--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit c280e36a84c5cb3ac6bb60e285b9c53676533a1a
Parents : 03d1b22
Author : RFNexus <v6z777@protonmail.com>
Date : 2025-03-09T15:20:36-04:00
Interface Management (1/2)
Changes
6 files changed, 3280 insertions(+), 13 deletions(-)
Diff
diff --git a/nomadnet/ui/TextUI.py b/nomadnet/ui/TextUI.py
index 3e4b3cf..11dd2c7 100644
--- a/nomadnet/ui/TextUI.py
+++ b/nomadnet/ui/TextUI.py
@@ -51,7 +51,15 @@ THEMES = {
("browser_controls", "light gray", "default", "default", "#bbb", "default"),
("progress_full", "black", "light gray", "standout", "#111", "#bbb"),
("progress_empty", "light gray", "default", "default", "#ddd", "default"),
- ],
+ ("interface_title", "", "", "default", "", ""),
+ ("interface_title_selected", "bold", "", "bold", "", ""),
+ ("connected_status", "dark green", "default", "default", "dark green", "default"),
+ ("disconnected_status", "dark red", "default", "default", "dark red", "default"),
+ ("placeholder", "dark gray", "default", "default", "dark gray", "default"),
+ ("placeholder_text", "dark gray", "default", "default", "dark gray", "default"),
+ ("error", "light red,blink", "default", "blink", "#f44,blink", "default"),
+
+ ],
},
THEME_LIGHT: {
"urwid_theme": [
@@ -69,6 +77,7 @@ THEMES = {
("msg_header_ok", "black", "dark green", "standout", "#111", "#6b2"),
("msg_header_caution", "black", "yellow", "standout", "#111", "#fd3"),
("msg_header_sent", "black", "dark gray", "standout", "#111", "#ddd"),
+ ("msg_header_propagated", "black", "light blue", "standout", "#111", "#28b"),
("msg_header_delivered", "black", "light blue", "standout", "#111", "#28b"),
("msg_header_failed", "black", "dark gray", "standout", "#000", "#777"),
("msg_warning_untrusted", "black", "dark red", "standout", "#111", "dark red"),
@@ -86,6 +95,13 @@ THEMES = {
("browser_controls", "dark gray", "default", "default", "#444", "default"),
("progress_full", "black", "dark gray", "standout", "#111", "#bbb"),
("progress_empty", "dark gray", "default", "default", "#ddd", "default"),
+ ("interface_title", "dark gray", "default", "default", "#444", "default"),
+ ("interface_title_selected", "dark gray,bold", "default", "bold", "#444,bold", "default"),
+ ("connected_status", "dark green", "default", "default", "#4a0", "default"),
+ ("disconnected_status", "dark red", "default", "default", "#a22", "default"),
+ ("placeholder", "light gray", "default", "default", "#999", "default"),
+ ("placeholder_text", "light gray", "default", "default", "#999", "default"),
+ ("error", "dark red,blink", "default", "blink", "#a22,blink", "default"),
],
}
}
@@ -128,6 +144,8 @@ GLYPHS = {
("sent", "/\\", "\u2191", "\U000f0cd8"),
("papermsg", "P", "\u25a4", "\uf719"),
("qrcode", "QR", "\u25a4", "\uf029"),
+ ("selected", "[*] ", "\u25CF", "\u25CF"),
+ ("unselected", "[ ] ", "\u25CB", "\u25CB"),
}
class TextUI:
@@ -163,7 +181,7 @@ class TextUI:
if self.app.config["textui"]["glyphs"] == "plain":
glyphset = "plain"
- elif self.app.config["textui"]["glyphs"] == "unicoode":
+ elif self.app.config["textui"]["glyphs"] == "unicode":
glyphset = "unicode"
elif self.app.config["textui"]["glyphs"] == "nerdfont":
glyphset = "nerdfont"
diff --git a/nomadnet/ui/textui/Guide.py b/nomadnet/ui/textui/Guide.py
index 6308edc..94f83d0 100644
--- a/nomadnet/ui/textui/Guide.py
+++ b/nomadnet/ui/textui/Guide.py
@@ -109,6 +109,7 @@ class TopicList(urwid.WidgetWrap):
self.topic_list = [
GuideEntry(self.app, self, guide_display, "Introduction"),
GuideEntry(self.app, self, guide_display, "Concepts & Terminology"),
+ GuideEntry(self.app, self, guide_display, "Interfaces"),
GuideEntry(self.app, self, guide_display, "Hosting a Node"),
GuideEntry(self.app, self, guide_display, "Configuration Options"),
GuideEntry(self.app, self, guide_display, "Keyboard Shortcuts"),
@@ -386,6 +387,10 @@ Links can be inserted into micron documents. See the `*Markup`* section of this
'''
+TOPIC_INTERFACES = '''
+>TODO
+'''
+
TOPIC_CONVERSATIONS = '''>Conversations
Conversations in Nomad Network
@@ -1247,6 +1252,7 @@ TOPICS = {
"Introduction": TOPIC_INTRODUCTION,
"Concepts & Terminology": TOPIC_CONCEPTS,
"Conversations": TOPIC_CONVERSATIONS,
+ "Interfaces": TOPIC_INTERFACES,
"Hosting a Node": TOPIC_HOSTING,
"Configuration Options": TOPIC_CONFIG,
"Keyboard Shortcuts": TOPIC_SHORTCUTS,
diff --git a/nomadnet/ui/textui/Interfaces.py b/nomadnet/ui/textui/Interfaces.py
new file mode 100644
index 0000000..281e505
--- /dev/null
+++ b/nomadnet/ui/textui/Interfaces.py
@@ -0,0 +1,2865 @@
+import RNS
+import time
+from math import log10, pow
+
+from nomadnet.vendor.additional_urwid_widgets.FormWidgets import *
+from nomadnet.vendor.AsciiChart import AsciiChart
+
+### GYLPHS ###
+INTERFACE_GLYPHS = {
+ # Glyph name # Plain # Unicode # Nerd Font
+ ("NetworkInterfaceType", "(IP)", "\U0001f5a7", "\U000f0200"),
+ ("SerialInterfaceType", "(<->)", "\u2194", "\U000f065c"),
+ ("RNodeInterfaceType", "(R)" , "\u16b1", "\U000f043a"),
+ ("OtherInterfaceType", "(#)" , "\U0001f67e", "\ued95"),
+}
+
+### HELPER ###
+def _get_interface_icon(glyphset, iface_type):
+ glyphset_index = 1 # Default to unicode
+ if glyphset == "plain":
+ glyphset_index = 0 # plain
+ elif glyphset == "nerdfont":
+ glyphset_index = 2 # nerdfont
+
+ type_to_glyph_tuple = {
+ "AutoInterface": "NetworkInterfaceType",
+ "TCPClientInterface": "NetworkInterfaceType",
+ "TCPServerInterface": "NetworkInterfaceType",
+ "UDPInterface": "NetworkInterfaceType",
+ "I2PInterface": "NetworkInterfaceType",
+
+ "RNodeInterface": "RNodeInterfaceType",
+ "RNodeMultiInterface": "RNodeInterfaceType",
+
+ "SerialInterface": "SerialInterfaceType",
+ "KISSInterface": "SerialInterfaceType",
+ "AX25KISSInterface": "SerialInterfaceType",
+
+ "PipeInterface": "OtherInterfaceType"
+ }
+
+ glyph_tuple_name = type_to_glyph_tuple.get(iface_type, "OtherInterfaceType")
+
+ for glyph_tuple in INTERFACE_GLYPHS:
+ if glyph_tuple[0] == glyph_tuple_name:
+ return glyph_tuple[glyphset_index + 1]
+
+ # Fallback
+ return "(#)" if glyphset == "plain" else "🙾" if glyphset == "unicode" else " "
+
+def format_bytes(bytes_value):
+ units = ['bytes', 'KB/s', 'MB/s', 'GB/s', 'TB/s']
+ size = float(bytes_value)
+ unit_index = 0
+
+ while size >= 1024.0 and unit_index < len(units) - 1:
+ size /= 1024.0
+ unit_index += 1
+
+ if unit_index == 0:
+ return f"{int(size)} {units[unit_index]}"
+ else:
+ return f"{size:.1f} {units[unit_index]}"
+
+### PORT FUNCTIONS ###
+PYSERIAL_AVAILABLE = False # If NomadNet is installed on environments with rnspure instead of rns, pyserial won't be available
+try:
+ import serial.tools.list_ports
+ PYSERIAL_AVAILABLE = True
+except ImportError:
+ class DummyPort:
+ def __init__(self, device, description=None, manufacturer=None, hwid=None):
+ self.device = device
+ self.description = description or device
+ self.manufacturer = manufacturer
+ self.hwid = hwid
+ self.vid = None
+ self.pid = None
+
+def get_port_info():
+ if not PYSERIAL_AVAILABLE:
+ return []
+
+ try:
+ ports = serial.tools.list_ports.comports()
+ port_info = []
+
+ # Ports are sorted into categories for dropdown, priority ports appear first
+ priority_ports = [] # USB, ACM, bluetooth, etc
+ standard_ports = [] # COM, tty/s ports
+
+ for port in ports:
+ desc = f"{port.device}"
+ if port.description and port.description != port.device:
+ desc += f" ({port.description})"
+ if port.manufacturer:
+ desc += f" - {port.manufacturer}"
+
+ is_standard = (
+ port.device.startswith("COM") or # windows
+ "/dev/ttyS" in port.device or # Linux
+ "Serial" in port.description
+ )
+
+ port_data = {
+ 'device': port.device,
+ 'description': desc,
+ 'hwid': port.hwid,
+ 'vid': port.vid,
+ 'pid': port.pid,
+ 'is_standard': is_standard
+ }
+
+ if is_standard:
+ standard_ports.append(port_data)
+ else:
+ priority_ports.append(port_data)
+
+ priority_ports.sort(key=lambda x: x['device'])
+ standard_ports.sort(key=lambda x: x['device'])
+
+ return priority_ports + standard_ports
+ except Exception as e:
+ RNS.log(f"error accessing serial ports: {str(e)}", RNS.LOG_ERROR)
+ return []
+
+def get_port_field():
+ if not PYSERIAL_AVAILABLE:
+ return {
+ "config_key": "port",
+ "type": "edit",
+ "label": "Port: ",
+ "default": "",
+ "placeholder": "/dev/ttyUSB0 or COM port (pyserial not installed)",
+ "validation": ["required"],
+ "transform": lambda x: x.strip()
+ }
+
+ port_info = get_port_info()
+
+ if len(port_info) > 1:
+ options = [p['description'] for p in port_info]
+ device_map = {p['description']: p['device'] for p in port_info}
+
+ return {
+ "config_key": "port",
+ "type": "dropdown",
+ "label": "Port: ",
+ "options": options,
+ "default": options[0] if options else "",
+ "validation": ["required"],
+ "transform": lambda x: device_map[x]
+ }
+ else:
+ # single or no ports - use text field
+ default = port_info[0]['device'] if port_info else ""
+ placeholder = "/dev/ttyXXX (or COM port on Windows)"
+
+ return {
+ "config_key": "port",
+ "type": "edit",
+ "label": "Port: ",
+ "default": default,
+ "placeholder": placeholder,
+ "validation": ["required"],
+ "transform": lambda x: x.strip()
+ }
+
+### RNODE ####
+def calculate_rnode_parameters(bandwidth, spreading_factor, coding_rate, noise_floor=6, antenna_gain=0,
+ transmit_power=17):
+ crn = {
+ 5: 1,
+ 6: 2,
+ 7: 3,
+ 8: 4,
+ }
+ coding_rate_n = crn.get(coding_rate, 1)
+
+ sfn = {
+ 5: -2.5,
+ 6: -5,
+ 7: -7.5,
+ 8: -10,
+ 9: -12.5,
+ 10: -15,
+ 11: -17.5,
+ 12: -20
+ }
+
+ data_rate = spreading_factor * (
+ (4 / (4 + coding_rate_n)) / (pow(2, spreading_factor) / (bandwidth / 1000))) * 1000
+
+ sensitivity = -174 + 10 * log10(bandwidth) + noise_floor + (sfn.get(spreading_factor, 0))
+
+ if bandwidth == 203125 or bandwidth == 406250 or bandwidth > 500000:
+ sensitivity = -165.6 + 10 * log10(bandwidth) + noise_floor + (sfn.get(spreading_factor, 0))
+
+ link_budget = (transmit_power - sensitivity) + antenna_gain
+
+ if data_rate < 1000:
+ data_rate_str = f"{data_rate:.0f} bps"
+ else:
+ data_rate_str = f"{(data_rate / 1000):.2f} kbps"
+
+ return {
+ "data_rate": data_rate_str,
+ "link_budget": f"{link_budget:.1f} dB",
+ "sensitivity": f"{sensitivity:.1f} dBm",
+ "raw_data_rate": data_rate,
+ "raw_link_budget": link_budget,
+ "raw_sensitivity": sensitivity
+ }
+
+class RNodeCalculator(urwid.WidgetWrap):
+ def __init__(self, parent_view):
+ self.parent_view = parent_view
+ self.update_alarm = None
+
+ self.data_rate_widget = urwid.Text("Data Rate: Calculating...")
+ self.link_budget_widget = urwid.Text("Link Budget: Calculating...")
+ self.sensitivity_widget = urwid.Text("Sensitivity: Calculating...")
+
+ self.noise_floor_edit = urwid.Edit("", "0")
+ self.antenna_gain_edit = urwid.Edit("", "0")
+
+ layout = urwid.Pile([
+ urwid.Divider("-"),
+
+ urwid.Columns([
+ (28, urwid.Text(("key", "Enter Noise Floor (dB): "), align="right")),
+ self.noise_floor_edit
+ ]),
+
+ urwid.Columns([
+ (28, urwid.Text(("key", "Enter Antenna Gain (dBi): "), align="right")),
+ self.antenna_gain_edit
+ ]),
+
+ urwid.Divider(),
+
+ urwid.Text(("connected_status", "On-Air Calculations:"), align="left"),
+ self.data_rate_widget,
+ self.link_budget_widget,
+ self.sensitivity_widget,
+ urwid.Divider(),
+
+ urwid.Text([
+ "These calculations will update as you change RNode parameters"
+ ])
+ ])
+
+ super().__init__(layout)
+
+ self.connect_all_field_signals()
+
+ self.update_calculation()
+
+ def connect_all_field_signals(self):
+ urwid.connect_signal(self.noise_floor_edit, 'change', self._queue_update)
+ urwid.connect_signal(self.antenna_gain_edit, 'change', self._queue_update)
+ rnode_fields = ['bandwidth', 'spreadingfactor', 'codingrate', 'txpower']
+
+ for field_name in rnode_fields:
+ if field_name in self.parent_view.fields:
+
+ field_widget = self.parent_view.fields[field_name]['widget']
+
+ if hasattr(field_widget, 'edit_text'):
+ urwid.connect_signal(field_widget, 'change', self._queue_update)
+ elif hasattr(field_widget, '_emit') and 'change' in getattr(field_widget, 'signals', []):
+ urwid.connect_signal(field_widget, 'change', self._queue_update)
+
+ def _queue_update(self, widget, new_text):
+ if self.update_alarm is not None:
+ try:
+ self.parent_view.parent.app.ui.loop.remove_alarm(self.update_alarm)
+ except:
+ pass
+
+ self.update_alarm = self.parent_view.parent.app.ui.loop.set_alarm_in(
+ 0.3, self._delayed_update)
+
+ def _delayed_update(self, loop, user_data):
+ self.update_alarm = None
+ self.update_calculation()
+
+ def update_calculation(self):
+ try:
+
+ try:
+ bandwidth_widget = self.parent_view.fields.get('bandwidth', {}).get('widget')
+ bandwidth = int(bandwidth_widget.get_value()) if bandwidth_widget else 125000
+ except (ValueError, AttributeError):
+ bandwidth = 125000
+
+ try:
+ sf_widget = self.parent_view.fields.get('spreadingfactor', {}).get('widget')
+ spreading_factor = int(sf_widget.get_value()) if sf_widget else 7
+ except (ValueError, AttributeError):
+ spreading_factor = 7
+
+ try:
+ cr_widget = self.parent_view.fields.get('codingrate', {}).get('widget')
+ coding_rate = int(cr_widget.get_value()) if cr_widget else 5
+ if isinstance(coding_rate, str) and ":" in coding_rate:
+ coding_rate = int(coding_rate.split(":")[1])
+ except (ValueError, AttributeError):
+ coding_rate = 5
+
+ try:
+ txpower_widget = self.parent_view.fields.get('txpower', {}).get('widget')
+ if hasattr(txpower_widget, 'edit_text'):
+ txpower_text = txpower_widget.edit_text.strip()
+ txpower = int(txpower_text) if txpower_text else 17
+ else:
+ txpower = int(txpower_widget.get_value()) if txpower_widget else 17
+ except (ValueError, AttributeError):
+ txpower = 17
+
+ try:
+ noise_floor_text = self.noise_floor_edit.edit_text.strip()
+ noise_floor = int(noise_floor_text) if noise_floor_text else 0
+ except (ValueError, AttributeError):
+ noise_floor = 0
+
+ try:
+ antenna_gain_text = self.antenna_gain_edit.edit_text.strip()
+ antenna_gain = int(antenna_gain_text) if antenna_gain_text else 0
+ except (ValueError, AttributeError):
+ antenna_gain = 0
+
+ result = calculate_rnode_parameters(
+ bandwidth=bandwidth,
+ spreading_factor=spreading_factor,
+ coding_rate=coding_rate,
+ noise_floor=noise_floor,
+ antenna_gain=antenna_gain,
+ transmit_power=txpower
+ )
+
+ self.data_rate_widget.set_text(f"Data Rate: {result['data_rate']}")
+ self.link_budget_widget.set_text(f"Link Budget: {result['link_budget']}")
+ self.sensitivity_widget.set_text(f"Sensitivity: {result['sensitivity']}")
+
+ except (ValueError, KeyError, TypeError) as e:
+ self.data_rate_widget.set_text(f"Data Rate: Waiting for parameters...")
+ self.link_budget_widget.set_text(f"Link Budget: Waiting for valid parameters...")
+ self.sensitivity_widget.set_text(f"Sensitivity: Waiting for parameters...")
+
+### INTERFACE FIELDS ###
+COMMON_INTERFACE_OPTIONS = [
+ {
+ "config_key": "network_name",
+ "type": "edit",
+ "label": "Virtual Network Name: ",
+ "placeholder": "Optional virtual network name",
+ "default": "",
+ "validation": [],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "passphrase",
+ "type": "edit",
+ "label": "IFAC Passphrase: ",
+ "placeholder": "IFAC authentication passphrase",
+ "default": "",
+ "validation": [],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "ifac_size",
+ "type": "edit",
+ "label": "IFAC Size: ",
+ "placeholder": "8 - 512",
+ "default": "",
+ "validation": ['number'],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "bitrate",
+ "type": "edit",
+ "label": "Inferred Bitrate: ",
+ "placeholder": "Automatically determined",
+ "default": "",
+ "validation": ['number'],
+ "transform": lambda x: x.strip()
+ },
+]
+
+INTERFACE_FIELDS = {
+ "AutoInterface": [
+ {
+
+ },
+ {
+ "additional_options": [
+ {
+ "config_key": "devices",
+ "type": "multilist",
+ "label": "Devices: ",
+ "validation": [],
+ "transform": lambda x: ",".join(x)
+ },
+ {
+ "config_key": "ignored_devices",
+ "type": "multilist",
+ "label": "Ignored Devices: ",
+ "validation": [],
+ "transform": lambda x: ",".join(x)
+ },
+ {
+ "config_key": "group_id",
+ "type": "edit",
+ "label": "Group ID: ",
+ "default": "",
+ "placeholder": "e.g., my_custom_network",
+ "validation": [],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "discovery_scope",
+ "type": "dropdown",
+ "label": "Discovery Scope: ",
+ "options": ["None", "link", "admin", "site", "organisation", "global"],
+ "default": "None",
+ "validation": [],
+ "transform": lambda x: "" if x == "None" else x.strip()
+ }
+ ]
+ },
+ ],
+ "I2PInterface": [
+ {
+ "config_key": "peers",
+ "type": "multilist",
+ "label": "Peers: ",
+ "placeholder": "",
+ "validation": ["required"],
+ "transform": lambda x: ",".join(x)
+ }
+ ],
+ "TCPServerInterface": [
+ {
+ "config_key": "listen_ip",
+ "type": "edit",
+ "label": "Listen IP: ",
+ "default": "",
+ "placeholder": "e.g., 0.0.0.0",
+ "validation": ["required"],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "listen_port",
+ "type": "edit",
+ "label": "Listen Port: ",
+ "default": "",
+ "placeholder": "e.g., 4242",
+ "validation": ["required", "number"],
+ "transform": lambda x: int(x.strip()) if x.strip() else 4242
+ },
+ {
+ "additional_options": [
+ {
+ "config_key": "prefer_ipv6",
+ "type": "checkbox",
+ "label": "Prefer IPv6?",
+ "default": False,
+ "validation": [],
+ "transform": lambda x: bool(x)
+ },
+ {
+ "config_key": "i2p_tunneled",
+ "type": "checkbox",
+ "label": "I2P Tunneled?",
+ "default": False,
+ "validation": [],
+ "transform": lambda x: bool(x)
+ },
+ {
+ "config_key": "device",
+ "type": "edit",
+ "label": "Device: ",
+ "placeholder": "A specific network device to listen on - e.g. eth0",
+ "default": "",
+ "validation": [],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "port",
+ "type": "edit",
+ "label": "Port: ",
+ "default": "",
+ "placeholder": "e.g., 4242",
+ "validation": ["number"],
+ "transform": lambda x: int(x.strip()) if x.strip() else 4242
+ },
+ ]
+ }
+ ],
+ "TCPClientInterface": [
+ {
+ "config_key": "target_host",
+ "type": "edit",
+ "label": "Target Host: ",
+ "default": "",
+ "placeholder": "e.g., 127.0.0.1",
+ "validation": ["required"],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "target_port",
+ "type": "edit",
+ "label": "Target Port: ",
+ "default": "",
+ "placeholder": "e.g., 8080",
+ "validation": ["required", "number"],
+ "transform": lambda x: int(x.strip()) if x.strip() else 4242
+ },
+ {
+ "additional_options": [
+ {
+ "config_key": "i2p_tunneled",
+ "type": "checkbox",
+ "label": "I2P Tunneled?",
+ "default": False,
+ "validation": [],
+ "transform": lambda x: bool(x)
+ },
+ {
+ "config_key": "kiss_framing",
+ "type": "checkbox",
+ "label": "KISS Framing?",
+ "default": False,
+ "validation": [],
+ "transform": lambda x: bool(x)
+ }
+ ]
+ }
+ ],
+ "UDPInterface": [
+ {
+ "config_key": "listen_ip",
+ "type": "edit",
+ "label": "Listen IP: ",
+ "default": "",
+ "placeholder": "e.g., 0.0.0.0",
+ "validation": ["required"],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "listen_port",
+ "type": "edit",
+ "label": "Listen Port: ",
+ "default": "",
+ "placeholder": "e.g., 4242",
+ "validation": ["required", "number"],
+ "transform": lambda x: int(x.strip()) if x.strip() else 4242
+ },
+ {
+ "config_key": "forward_ip",
+ "type": "edit",
+ "label": "Forward IP: ",
+ "default": "",
+ "placeholder": "e.g., 255.255.255.255",
+ "validation": ["required"],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "forward_port",
+ "type": "edit",
+ "label": "Forward Port: ",
+ "default": "",
+ "placeholder": "e.g., 4242",
+ "validation": ["required", "number"],
+ "transform": lambda x: int(x.strip()) if x.strip() else 4242
+ },
+ {
+ "additional_options": [
+ {
+ "config_key": "device",
+ "type": "edit",
+ "label": "Device: ",
+ "placeholder": "A specific network device to listen on - e.g. eth0",
+ "default": "",
+ "validation": [],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "port",
+ "type": "edit",
+ "label": "Port: ",
+ "default": "",
+ "placeholder": "e.g., 4242",
+ "validation": ["number"],
+ "transform": lambda x: int(x.strip()) if x.strip() else 4242
+ },
+ ]
+ }
+ ],
+ "RNodeInterface": [
+ get_port_field(),
+ {
+ "config_key": "frequency",
+ "type": "edit",
+ "label": "Frequency (MHz): ",
+ "default": "",
+ "placeholder": "868.5",
+ "validation": ["required", "float"],
+ "transform": lambda x: int(float(x.strip()) * 1000000) if x.strip() else 868500000
+ },
+ {
+ "config_key": "txpower",
+ "type": "edit",
+ "label": "Transmit Power (dBm): ",
+ "default": "",
+ "placeholder": "17",
+ "validation": ["required", "number"],
+ "transform": lambda x: int(x.strip()) if x.strip() else 17
+ },
+ {
+ "config_key": "bandwidth",
+ "type": "dropdown",
+ "label": "Bandwidth (Hz): ",
+ "options": ["7800", "10400", "15600", "20800", "31250", "41700", "62500", "125000", "250000",
+ "500000", "1625000"],
+ "default": "7800",
+ "validation": ["required"],
+ "transform": lambda x: int(x)
+ },
+ {
+ "config_key": "spreadingfactor",
+ "type": "dropdown",
+ "label": "Spreading Factor: ",
+ "options": ["7", "8", "9", "10", "11", "12"],
+ "default": "7",
+ "validation": ["required"],
+ "transform": lambda x: int(x)
+ },
+ {
+ "config_key": "codingrate",
+ "type": "dropdown",
+ "label": "Coding Rate: ",
+ "options": ["4:5", "4:6", "4:7", "4:8"],
+ "default": "4:5",
+ "validation": ["required"],
+ "transform": lambda x: int(x.split(":")[1])
+ },
+ {
+ "additional_options": [
+ {
+ "config_key": "id_callsign",
+ "type": "edit",
+ "label": "Callsign: ",
+ "default": "",
+ "placeholder": "e.g. MYCALL-0",
+ "validation": [""],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "id_interval",
+ "type": "edit",
+ "label": "ID Interval (Seconds): ",
+ "placeholder": "e.g. 600",
+ "default": "",
+ "validation": ['number'],
+ "transform": lambda x: "" if x == "" else int(x)
+ },
+ {
+ "config_key": "airtime_limit_long",
+ "type": "edit",
+ "label": "Airtime Limit Long (Seconds): ",
+ "placeholder": "e.g. 1.5",
+ "default": "",
+ "validation": ['number'],
+ "transform": lambda x: "" if x == "" else int(x)
+ },
+ {
+ "config_key": "airtime_limit_short",
+ "type": "edit",
+ "label": "Airtime Limit Short (Seconds): ",
+ "placeholder": "e.g. 33",
+ "default": "",
+ "validation": ['number'],
+ "transform": lambda x: "" if x == "" else int(x)
+ },
+ ]
+ }
+ ],
+ "SerialInterface": [
+ get_port_field(),
+ {
+ "config_key": "speed",
+ "type": "edit",
+ "label": "Speed (bps): ",
+ "default": "",
+ "placeholder": "e.g. 115200",
+ "validation": ["required", "number"],
+ "transform": lambda x: int(x.strip())
+ },
+ {
+ "config_key": "databits",
+ "type": "edit",
+ "label": "Databits: ",
+ "default": "",
+ "placeholder": "e.g. 8",
+ "validation": ["required", "number"],
+ "transform": lambda x: int(x.strip())
+ },
+ {
+ "config_key": "parity",
+ "type": "edit",
+ "label": "Parity: ",
+ "default": "",
+ "placeholder": "",
+ "validation": ["number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "config_key": "stopbits",
+ "type": "edit",
+ "label": "Stopbits: ",
+ "default": "",
+ "placeholder": "e.g. 1",
+ "validation": ["number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ ],
+ "PipeInterface": [
+ {
+ "config_key": "command",
+ "type": "edit",
+ "label": "Command: ",
+ "default": "",
+ "placeholder": "e.g. netcat -l 5757",
+ "validation": ["required"],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "respawn_delay",
+ "type": "edit",
+ "label": "Respawn Delay (seconds): ",
+ "default": "",
+ "placeholder": "e.g. 5",
+ "validation": ["number"],
+ "transform": lambda x: x.strip()
+ },
+ ],
+ "KISSInterface": [
+ get_port_field(),
+ {
+ "config_key": "speed",
+ "type": "edit",
+ "label": "Speed (bps): ",
+ "default": "",
+ "placeholder": "e.g. 115200",
+ "validation": ["required", "number"],
+ "transform": lambda x: int(x.strip())
+ },
+ {
+ "config_key": "databits",
+ "type": "edit",
+ "label": "Databits: ",
+ "default": "",
+ "placeholder": "e.g. 8",
+ "validation": ["required", "number"],
+ "transform": lambda x: int(x.strip())
+ },
+ {
+ "config_key": "parity",
+ "type": "edit",
+ "label": "Parity: ",
+ "default": "",
+ "placeholder": "",
+ "validation": ["number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "config_key": "stopbits",
+ "type": "edit",
+ "label": "Stopbits: ",
+ "default": "",
+ "placeholder": "e.g. 1",
+ "validation": ["number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "config_key": "preamble",
+ "type": "edit",
+ "label": "Preamble (miliseconds): ",
+ "default": "",
+ "placeholder": "e.g. 150",
+ "validation": ["required", "number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "config_key": "txtail",
+ "type": "edit",
+ "label": "TX Tail (miliseconds): ",
+ "default": "",
+ "placeholder": "e.g. 10",
+ "validation": ["required", "number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "config_key": "slottime",
+ "type": "edit",
+ "label": "slottime (miliseconds): ",
+ "default": "",
+ "placeholder": "e.g. 20",
+ "validation": ["required", "number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "config_key": "persistence",
+ "type": "edit",
+ "label": "Persistence (miliseconds): ",
+ "default": "",
+ "placeholder": "e.g. 200",
+ "validation": ["required", "number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "additional_options": [
+ {
+ "config_key": "id_callsign",
+ "type": "edit",
+ "label": "ID Callsign: ",
+ "default": "",
+ "placeholder": "e.g. MYCALL-0",
+ "validation": [""],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "id_interval",
+ "type": "edit",
+ "label": "ID Interval (Seconds): ",
+ "placeholder": "e.g. 600",
+ "default": "",
+ "validation": ['number'],
+ "transform": lambda x: "" if x == "" else int(x)
+ },
+ {
+ "config_key": "flow_control",
+ "type": "checkbox",
+ "label": "Flow Control ",
+ "validation": [],
+ "transform": lambda x: "" if x == "" else bool(x)
+ },
+ ]
+ }
+ ],
+ "AX25KISSInterface": [
+ get_port_field(),
+ {
+ "config_key": "callsign",
+ "type": "edit",
+ "label": "Callsign: ",
+ "default": "",
+ "placeholder": "e.g. NO1CLL",
+ "validation": ["required"],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "ssid",
+ "type": "edit",
+ "label": "SSID: ",
+ "default": "",
+ "placeholder": "e.g. 0",
+ "validation": ["required"],
+ "transform": lambda x: x.strip()
+ },
+ {
+ "config_key": "speed",
+ "type": "edit",
+ "label": "Speed (bps): ",
+ "default": "",
+ "placeholder": "e.g. 115200",
+ "validation": ["required", "number"],
+ "transform": lambda x: int(x.strip())
+ },
+ {
+ "config_key": "databits",
+ "type": "edit",
+ "label": "Databits: ",
+ "default": "",
+ "placeholder": "e.g. 8",
+ "validation": ["required", "number"],
+ "transform": lambda x: int(x.strip())
+ },
+ {
+ "config_key": "parity",
+ "type": "edit",
+ "label": "Parity: ",
+ "default": "",
+ "placeholder": "",
+ "validation": ["number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "config_key": "stopbits",
+ "type": "edit",
+ "label": "Stopbits: ",
+ "default": "",
+ "placeholder": "e.g. 1",
+ "validation": ["number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "config_key": "preamble",
+ "type": "edit",
+ "label": "Preamble (miliseconds): ",
+ "default": "",
+ "placeholder": "e.g. 150",
+ "validation": ["required", "number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "config_key": "txtail",
+ "type": "edit",
+ "label": "TX Tail (miliseconds): ",
+ "default": "",
+ "placeholder": "e.g. 10",
+ "validation": ["required", "number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "config_key": "slottime",
+ "type": "edit",
+ "label": "Slottime (miliseconds): ",
+ "default": "",
+ "placeholder": "e.g. 20",
+ "validation": ["required", "number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "config_key": "persistence",
+ "type": "edit",
+ "label": "Persistence (miliseconds): ",
+ "default": "",
+ "placeholder": "e.g. 200",
+ "validation": ["required", "number"],
+ "transform": lambda x: "" if x == "" else int(x.strip())
+ },
+ {
+ "additional_options": [
+ {
+ "config_key": "flow_control",
+ "type": "checkbox",
+ "label": "Flow Control ",
+ "validation": [],
+ "transform": lambda x: "" if x == "" else bool(x)
+ },
+ ]
+ }
+ ],
+ "default": [
+ {
+
+ },
+ ]
+}
+
+### INTERFACE WIDGETS ####
+class SelectableInterfaceItem(urwid.WidgetWrap):
+ def __init__(self, parent, name, is_connected, is_enabled, iface_type, tx, rx, icon="?", iface_options=None):
+ self.parent = parent
+ self._selectable = True
+ self.icon = icon
+ self.name = name
+ self.is_connected = is_connected
+ self.is_enabled = is_enabled
+ self.iface_options = iface_options
+
+
+ if is_enabled:
+ enabled_txt = ("connected_status", "Enabled")
+ else:
+ enabled_txt = ("disconnected_status", "Disabled")
+
+ if is_connected:
+ connected_txt = ("connected_status", "Connected")
+ else:
+ connected_txt = ("disconnected_status", "Disconnected")
+
+ self.selection_txt = urwid.Text(" ")
+ self.title_widget = urwid.Text(("interface_title", f"{icon} {name}"))
+
+ title_content = urwid.Columns([
+ (4, self.selection_txt),
+ self.title_widget,
+ ])
+
+ self.tx_widget = urwid.Text(("value", format_bytes(tx)))
+ self.rx_widget = urwid.Text(("value", format_bytes(rx)))
+
+ self.status_widget = urwid.Text(enabled_txt)
+ self.connection_widget = urwid.Text(connected_txt)
+
+ rows = [
+ urwid.Columns([
+ (10, urwid.Text(("key", "Status: "))),
+ (10, self.status_widget),
+ (3, urwid.Text(" | ")),
+ self.connection_widget,
+ ]),
+
+ urwid.Columns([
+ (10, urwid.Text(("key", "Type:"))),
+ urwid.Text(("value", iface_type)),
+ ]),
+
+ urwid.Divider("-"),
+
+ urwid.Columns([
+ (10, urwid.Text(("key", "TX:"))),
+ (15, self.tx_widget),
+ (10, urwid.Text(("key", "RX:"))),
+ self.rx_widget,
+ ]),
+ ]
+
+ pile_contents = [title_content] + rows
+
+ pile = urwid.Pile(pile_contents)
+
+ padded_body = urwid.Padding(pile, left=2, right=2)
+
+ box = urwid.LineBox(
+ padded_body,
+ title=None,
+ #todo
+ tlcorner="╭", tline="─",
+ trcorner="╮", lline="│",
+ rline="│", blcorner="╰",
+ bline="─", brcorner="╯"
+ )
+
+ super().__init__(box)
+
+ def update_status_display(self):
+ if self.is_enabled:
+ self.status_widget.set_text(("connected_status", "Enabled"))
+ else:
+ self.status_widget.set_text(("disconnected_status", "Disabled"))
+
+ def selectable(self):
+ return True
+
+ def render(self, size, focus=False):
+ self.selection_txt.set_text(self.parent.g['selected'] if focus else self.parent.g['unselected'])
+
+ if focus:
+ self.title_widget.set_text(
+ ("interface_title_selected", f"{self.icon} {self.name}"))
+ else:
+ self.title_widget.set_text(("interface_title", f"{self.icon} {self.name}"))
+
+ return super().render(size, focus=focus)
+
+ def keypress(self, size, key):
+ if key == "enter":
+ self.parent.switch_to_show_interface(self.name)
+ return None
+ return key
+
+ def update_stats(self, tx, rx):
+ self.tx_widget.set_text(("value", format_bytes(tx)))
+ self.rx_widget.set_text(("value", format_bytes(rx)))
+
+class InterfaceOptionItem(urwid.WidgetWrap):
+ def __init__(self, parent_display, label, value):
+ self.parent_display = parent_display
+ self.label = label
+ self.value = value
+ self._selectable = True
+
+ text_widget = urwid.Text(label, align="left")
+ super().__init__(urwid.AttrMap(text_widget, "list_normal", focus_map="list_focus"))
+
+ def selectable(self):
+ return True
+
+ def keypress(self, size, key):
+ if key == "enter":
+ self.parent_display.dismiss_dialog()
+ self.parent_display.switch_to_add_interface(self.value)
+ return None
+ return super().keypress(size, key)
+
+class InterfaceBandwidthChart:
+
+ def __init__(self, history_length=60, glyphset="unicode"):
+ self.history_length = history_length
+ self.glyphset = glyphset
+ self.rx_rates = [0] * history_length
+ self.tx_rates = [0] * history_length
+
+ self.prev_rx = None
+ self.prev_tx = None
+ self.prev_time = None
+
+ self.max_rx_rate = 1
+ self.max_tx_rate = 1
+
+ self.first_update = True
+ self.initialization_complete = False
+ self.stabilization_updates = 2
+ self.update_count = 0
+
+ self.peak_rx_for_display = 0
+ self.peak_tx_for_display = 0
+
+ def update(self, rx_bytes, tx_bytes):
+ current_time = time.time()
+
+ if self.prev_rx is None or self.first_update:
+ self.prev_rx = rx_bytes
+ self.prev_tx = tx_bytes
+ self.prev_time = current_time
+ self.first_update = False
+ return
+
+ time_delta = max(0.1, current_time - self.prev_time)
+
+ rx_delta = max(0, rx_bytes - self.prev_rx) / time_delta
+ tx_delta = max(0, tx_bytes - self.prev_tx) / time_delta
+
+ self.prev_rx = rx_bytes
+ self.prev_tx = tx_bytes
+ self.prev_time = current_time
+
+ self.update_count += 1
+
+ self.rx_rates.pop(0)
+ self.tx_rates.pop(0)
+ self.rx_rates.append(rx_delta)
+ self.tx_rates.append(tx_delta)
+
+ if self.update_count >= self.stabilization_updates:
+ self.initialization_complete = True
+
+ self.peak_rx_for_display = max(self.peak_rx_for_display, rx_delta)
+ self.peak_tx_for_display = max(self.peak_tx_for_display, tx_delta)
+
+
+ current_rx_max = max(self.rx_rates)
+ current_tx_max = max(self.tx_rates)
+
+ self.max_rx_rate = max(1, current_rx_max)
+ self.max_tx_rate = max(1, current_tx_max)
+
+ def get_charts(self, height=8):
+ chart = AsciiChart(glyphset=self.glyphset)
+
+ rx_data = self.rx_rates.copy()
+ tx_data = self.tx_rates.copy()
+
+ peak_rx = self.peak_rx_for_display if self.initialization_complete else 0
+ peak_tx = self.peak_tx_for_display if self.initialization_complete else 0
+
+ peak_rx_str = format_bytes(peak_rx)
+ peak_tx_str = format_bytes(peak_tx)
+
+ rx_chart = chart.plot(
+ [rx_data],
+ {
+ 'height': height,
+ 'format': '{:8.1f}',
+ 'min': 0,
+ 'max': self.max_rx_rate * 1.1,
+ }
+ )
+
+ tx_chart = chart.plot(
+ [tx_data],
+ {
+ 'height': height,
+ 'format': '{:8.1f}',
+ 'min': 0,
+ 'max': self.max_tx_rate * 1.1,
+ }
+ )
+
+ return rx_chart, tx_chart, peak_rx_str, peak_tx_str
+
+
+class ResponsiveChartContainer(urwid.WidgetWrap):
+
+ def __init__(self, rx_box, tx_box, min_cols_for_horizontal=100):
+ self.rx_box = rx_box
+ self.tx_box = tx_box
+ self.min_cols_for_horizontal = min_cols_for_horizontal
+
+ self.horizontal_layout = urwid.Columns([
+ (urwid.WEIGHT, 1, self.rx_box),
+ (urwid.WEIGHT, 1, self.tx_box)
+ ])
+
+ self.vertical_layout = urwid.Pile([
+ self.rx_box,
+ self.tx_box
+ ])
+
+ self.layout = urwid.WidgetPlaceholder(self.horizontal_layout)
+
+ super().__init__(self.layout)
+
+ def render(self, size, focus=False):
+ maxcol = size[0] if len(size) > 0 else 0
+
+ if maxcol >= self.min_cols_for_horizontal and self.layout.original_widget is not self.horizontal_layout:
+ self.layout.original_widget = self.horizontal_layout
+ elif maxcol < self.min_cols_for_horizontal and self.layout.original_widget is not self.vertical_layout:
+ self.layout.original_widget = self.vertical_layout
+
+ return super().render(size, focus)
+
+### URWID FILLER ###
+class InterfaceFiller(urwid.WidgetWrap):
+ def __init__(self, widget, app):
+ self.app = app
+ self.filler = urwid.Filler(widget, urwid.TOP)
+ super().__init__(self.filler)
+
+ def keypress(self, size, key):
+ if key == "ctrl a":
+ # add interface
+ self.app.ui.main_display.sub_displays.interface_display.add_interface()
+ return
+ elif key == "ctrl x":
+ # remove Interface
+ self.app.ui.main_display.sub_displays.interface_display.remove_selected_interface()
+ return
+ elif key == "ctrl e":
+ # edit interface
+ self.app.ui.main_display.sub_displays.interface_display.edit_selected_interface()
+ return None
+ elif key == "tab":
+ # navigation
+ self.app.ui.main_display.frame.focus_position = "header"
+ return
+
+ return super().keypress(size, key)
+
+### VIEWS ###
+class AddInterfaceView(urwid.WidgetWrap):
+ def __init__(self, parent, iface_type):
+ self.parent = parent
+ self.iface_type = iface_type
+ self.fields = {}
+ self.port_pile = None
+ self.additional_fields = {}
+ self.additional_pile_contents = []
+ self.common_fields = {}
+
+ self.parent.shortcuts_display.set_add_interface_shortcuts()
+
+ name_field = FormEdit(
+ config_key="name",
+ placeholder="Enter interface name",
+ validation_types=["required"]
+ )
+ self.fields['name'] = {
+ 'label': "Name: ",
+ 'widget': name_field
+ }
+
+ config = INTERFACE_FIELDS.get(iface_type, INTERFACE_FIELDS["default"])
+ iface_fields = [field for field in config if "config_key" in field]
+
+ for field in iface_fields:
+ self._initialize_field(field)
+
+ self._initialize_additional_fields(config)
+
+ self._initialize_common_fields()
+
+ pile_items = self._build_form_layout(iface_fields)
+
+ form_pile = urwid.Pile(pile_items)
+ form_filler = urwid.Filler(form_pile, valign="top")
+ #todo
+ form_box = urwid.LineBox(
+ form_filler,
+ title="Add Interface",
+ tlcorner="╭", tline="─",
+ trcorner="╮", lline="│",
+ rline="│", blcorner="╰",
+ bline="─", brcorner="╯"
+ )
+
+ background = urwid.SolidFill(" ")
+ self.overlay = urwid.Overlay(
+ top_w=form_box,
+ bottom_w=background,
+ align='center',
+ width=('relative', 85),
+ valign='middle',
+ height=('relative', 85),
+
+ )
+ super().__init__(self.overlay)
+
+ def _initialize_field(self, field):
+ if field["type"] == "dropdown":
+ widget = FormDropdown(
+ config_key=field["config_key"],
+ label=field["label"],
+ options=field["options"],
+ default=field.get("default"),
+ validation_types=field.get("validation", []),
+ transform=field.get("transform")
+ )
+ elif field["type"] == "checkbox":
+ widget = FormCheckbox(
+ config_key=field["config_key"],
+ label=field["label"],
+ state=field.get("default", False),
+ validation_types=field.get("validation", []),
+ transform=field.get("transform")
+ )
+ elif field["type"] == "multilist":
+ widget = FormMultiList(
+ config_key=field["config_key"],
+ placeholder=field.get("placeholder", ""),
+ validation_types=field.get("validation", []),
+ transform=field.get("transform")
+ )
+ else:
+ widget = FormEdit(
+ config_key=field["config_key"],
+ caption="",
+ edit_text=field.get("default", ""),
+ placeholder=field.get("placeholder", ""),
+ validation_types=field.get("validation", []),
+ transform=field.get("transform")
+ )
+
+ self.fields[field["config_key"]] = {
+ 'label': field["label"],
+ 'widget': widget
+ }
+
+ def _initialize_additional_fields(self, config):
+ for field in config:
+ if isinstance(field, dict) and "additional_options" in field:
+ for option in field["additional_options"]:
+ if option["type"] == "checkbox":
+ widget = FormCheckbox(
+ config_key=option["config_key"],
+ label=option["label"],
+ state=option.get("default", False),
+ validation_types=option.get("validation", []),
+ transform=option.get("transform")
+ )
+ elif option["type"] == "dropdown":
+ widget = FormDropdown(
+ config_key=option["config_key"],
+ label=option["label"],
+ options=option["options"],
+ default=option.get("default"),
+ validation_types=option.get("validation", []),
+ transform=option.get("transform")
+ )
+ elif option["type"] == "multilist":
+ widget = FormMultiList(
+ config_key=option["config_key"],
+ placeholder=option.get("placeholder", ""),
+ validation_types=option.get("validation", []),
+ transform=option.get("transform")
+ )
+ else:
+ widget = FormEdit(
+ config_key=option["config_key"],
+ caption="",
+ edit_text=str(option.get("default", "")),
+ placeholder=option.get("placeholder", ""),
+ validation_types=option.get("validation", []),
+ transform=option.get("transform")
+ )
+
+ self.additional_fields[option["config_key"]] = {
+ 'label': option["label"],
+ 'widget': widget,
+ 'type': option["type"]
+ }
+
+ def _initialize_common_fields(self):
+
+ if self.parent.app.rns.transport_enabled():
+ # Transport mode options
+ COMMON_INTERFACE_OPTIONS.extend([
+ {
+ "config_key": "outgoing",
+ "type": "checkbox",
+ "label": "Allow outgoing traffic",
+ "default": True,
+ "validation": [],
+ "transform": lambda x: bool(x)
+ },
+ {
+ "config_key": "mode",
+ "type": "dropdown",
+ "label": "Interface Mode: ",
+ "options": ["full", "gateway", "access_point", "roaming", "boundary"],
+ "default": "full",
+ "validation": [],
+ "transform": lambda x: x
+ },
+ {
+ "config_key": "announce_cap",
+ "type": "edit",
+ "label": "Announce Cap: ",
+ "placeholder": "Default: 2.0",
+ "default": "",
+ "validation": ["float"],
+ "transform": lambda x: float(x) if x.strip() else 2.0
+ }
+ ])
+
+ for option in COMMON_INTERFACE_OPTIONS:
+ if option["type"] == "checkbox":
+ widget = FormCheckbox(
+ config_key=option["config_key"],
+ label=option["label"],
+ state=option.get("default", False),
+ validation_types=option.get("validation", []),
+ transform=option.get("transform")
+ )
+ elif option["type"] == "dropdown":
+ widget = FormDropdown(
+ config_key=option["config_key"],
+ label=option["label"],
+ options=option["options"],
+ default=option.get("default"),
+ validation_types=option.get("validation", []),
+ transform=option.get("transform")
+ )
+ else:
+ widget = FormEdit(
+ config_key=option["config_key"],
+ caption="",
+ edit_text=str(option.get("default", "")),
+ placeholder=option.get("placeholder", ""),
+ validation_types=option.get("validation", []),
+ transform=option.get("transform")
+ )
+
+
+
+ self.common_fields[option["config_key"]] = {
+ 'label': option["label"],
+ 'widget': widget,
+ 'type': option["type"]
+ }
+
+ def _on_rnode_field_change(self, widget, new_value):
+ if hasattr(self, 'rnode_calculator') and self.calculator_visible:
+ self.rnode_calculator.update_calculation()
+
+ def _build_form_layout(self, iface_fields):
+ pile_items = []
+ pile_items.append(urwid.Text(("form_title", f"Add new {_get_interface_icon(self.parent.glyphset, self.iface_type)} {self.iface_type}"), align="center"))
+ pile_items.append(urwid.Divider("─"))
+
+ for key in ["name"] + [f["config_key"] for f in iface_fields]:
+ field = self.fields[key]
+ widget = field["widget"]
+
+ field_pile = urwid.Pile([
+ urwid.Columns([
+ (26, urwid.Text(("key", field["label"]), align="right")),
+ widget,
+ ]),
+ urwid.Padding(widget.error_widget, left=24)
+ ])
+
+ if self.iface_type in ["RNodeInterface", "SerialInterface", "AX25KISSInterface", "KISSInterface"] and key == "port":
+ refresh_btn = urwid.Button("Refresh Ports", on_press=self.refresh_ports)
+ refresh_btn = urwid.AttrMap(refresh_btn, "button_normal", focus_map="button_focus")
+ refresh_row = urwid.Padding(refresh_btn, left=26, width=20)
+ field_pile.contents.append((refresh_row, field_pile.options()))
+ self.port_pile = field_pile
+
+ pile_items.append(field_pile)
+
+ self.more_options_visible = False
+ self.more_options_button = urwid.Button("Show more options", on_press=self.toggle_more_options)
+ self.more_options_button = urwid.AttrMap(self.more_options_button, "button_normal", focus_map="button_focus")
+ self.more_options_widget = urwid.Pile([])
+
+ self.ifac_options_visible = False
+ self.ifac_options_button = urwid.Button("Show IFAC options", on_press=self.toggle_ifac_options)
+ self.ifac_options_button = urwid.AttrMap(self.ifac_options_button, "button_normal", focus_map="button_focus")
+ self.ifac_options_widget = urwid.Pile([])
+
+ if self.iface_type == "RNodeInterface":
+ self.calculator_button = urwid.Button("Show On-Air Calculations", on_press=self.toggle_calculator)
+ self.calculator_button = urwid.AttrMap(self.calculator_button, "button_normal", focus_map="button_focus")
+
+ save_btn = urwid.Button("Save", on_press=self.on_save)
+ back_btn = urwid.Button("Cancel", on_press=self.on_back)
+ button_row = urwid.Columns([
+ (urwid.WEIGHT, 0.45, save_btn),
+ (urwid.WEIGHT, 0.1, urwid.Text("")),
+ (urwid.WEIGHT, 0.45, back_btn),
+ ])
+
+ pile_items.extend([
+ urwid.Divider(),
+ self.more_options_button,
+ self.more_options_widget,
+ ])
+ if self.iface_type == "RNodeInterface":
+ self.rnode_calculator = RNodeCalculator(self)
+ self.calculator_visible = False
+ self.calculator_widget = urwid.Pile([])
+ pile_items.extend([
+ self.calculator_button,
+ self.calculator_widget,
+ ])
+ pile_items.extend([
+ urwid.Divider("─"),
+ button_row,
+ ])
+
+ return pile_items
+
+ def toggle_more_options(self, button):
+ if self.more_options_visible:
+ self.more_options_widget.contents = []
+ button.base_widget.set_label("Show more options")
+ self.more_options_visible = False
+ else:
+ pile_contents = []
+
+ if self.additional_fields:
+ for key, field in self.additional_fields.items():
+ widget = field['widget']
+
+ if field['type'] == "checkbox":
+ centered_widget = urwid.Columns([
+ ('weight', 1, urwid.Text("")),
+ ('pack', widget),
+ ('weight', 1, urwid.Text(""))
+ ])
+ field_pile = urwid.Pile([
+ centered_widget,
+ urwid.Padding(widget.error_widget, left=24)
+ ])
+ else:
+ field_pile = urwid.Pile([
+ urwid.Columns([
+ (26, urwid.Text(("key", field["label"]), align="right")),
+ widget
+ ]),
+ urwid.Padding(widget.error_widget, left=24)
+ ])
+
+ pile_contents.append(field_pile)
+
+ if self.additional_fields and self.common_fields:
+ pile_contents.append(urwid.Divider("─"))
+
+ if self.common_fields:
+ # pile_contents.append(urwid.Text(("interface_title", "Common "), align="center"))
+ for key, field in self.common_fields.items():
+ widget = field['widget']
+
+ if field['type'] == "checkbox":
+ centered_widget = urwid.Columns([
+ ('weight', 1, urwid.Text("")),
+ ('pack', widget),
+ ('weight', 1, urwid.Text(""))
+ ])
+ field_pile = urwid.Pile([
+ centered_widget,
+ urwid.Padding(widget.error_widget, left=24)
+ ])
+ else:
+ field_pile = urwid.Pile([
+ urwid.Columns([
+ (26, urwid.Text(("key", field["label"]), align="right")),
+ widget
+ ]),
+ urwid.Padding(widget.error_widget, left=24)
+ ])
+
+ pile_contents.append(field_pile)
+
+ if pile_contents:
+ self.more_options_widget.contents = [(w, self.more_options_widget.options()) for w in pile_contents]
+ else:
+ self.more_options_widget.contents = [(
+ urwid.Text("No additional options available", align="center"),
+ self.more_options_widget.options()
+ )]
+
+ button.base_widget.set_label("Hide more options")
+ self.more_options_visible = True
+
+ def toggle_ifac_options(self, button):
+ if self.ifac_options_visible:
+ self.ifac_options_widget.contents = []
+ button.base_widget.set_label("Show IFAC options")
+ self.ifac_options_visible = False
+ else:
+ dummy = urwid.Text("IFAC (Interface Access Codes)", align="left")
+ self.ifac_options_widget.contents = [(dummy, self.more_options_widget.options())]
+ button.base_widget.set_label("Hide IFAC options")
+ self.ifac_options_visible = True
+
+ def toggle_calculator(self, button):
+ if self.calculator_visible:
+ self.calculator_widget.contents = []
+ button.base_widget.set_label("Show On-Air Calculations")
+ self.calculator_visible = False
+ else:
+ calculator_contents = [self.rnode_calculator]
+
+ self.calculator_widget.contents = [(w, self.calculator_widget.options()) for w in calculator_contents]
+
+ self.rnode_calculator.update_calculation()
+
+ button.base_widget.set_label("Hide On-Air Calculations")
+ self.calculator_visible = True
+
+ def refresh_ports(self, button):
+ if self.port_pile is not None:
+ # Get fresh port config
+ port_field = get_port_field()
+
+ if port_field["type"] == "dropdown":
+ widget = FormDropdown(
+ config_key=port_field["config_key"],
+ label=port_field["label"],
+ options=port_field["options"],
+ default=port_field.get("default"),
+ validation_types=port_field.get("validation", []),
+ transform=port_field.get("transform")
+ )
+ else:
+ widget = FormEdit(
+ config_key=port_field["config_key"],
+ caption="",
+ edit_text=port_field.get("default", ""),
+ placeholder=port_field.get("placeholder", ""),
+ validation_types=port_field.get("validation", []),
+ transform=port_field.get("transform")
+ )
+
+ self.fields["port"] = {
+ 'label': port_field["label"],
+ 'widget': widget
+ }
+
+ columns = urwid.Columns([
+ (26, urwid.Text(("key", port_field["label"]), align="right")),
+ widget
+ ])
+ self.port_pile.contents[0] = (columns, self.port_pile.options())
+
+ self.port_pile.contents[1] = (urwid.Padding(widget.error_widget, left=24), self.port_pile.options())
+
+ def validate_all(self):
+ all_valid = True
+
+ # validate main fields
+ for field in self.fields.values():
+ if not field["widget"].validate():
+ all_valid = False
+
+ # validate additional iface fields
+ for field in self.additional_fields.values():
+ if not field["widget"].validate():
+ all_valid = False
+
+ # validate common fields
+ for field in self.common_fields.values():
+ if not field["widget"].validate():
+ all_valid = False
+
+ return all_valid
+
+ def on_save(self, button):
+ all_valid = self.validate_all()
+
+ name = self.fields['name']["widget"].get_value() or "Untitled interface"
+
+ existing_interfaces = self.parent.app.rns.config['interfaces']
+ if name in existing_interfaces:
+ self.fields['name']["widget"].error = f"Interface name '{name}' already exists"
+ self.fields['name']["widget"].error_widget.set_text(("error", self.fields['name']["widget"].error))
+ all_valid = False
+
+ if not all_valid:
+ return
+
+ interface_config = {
+ "type": self.iface_type,
+ "interface_enabled": True
+ }
+
+ for field_key, field in self.fields.items():
+ if field_key != "name":
+ widget = field["widget"]
+ value = widget.get_value()
+ if value is not None and value != "":
+ interface_config[widget.config_key] = value
+
+ for field_key, field in self.additional_fields.items():
+ widget = field["widget"]
+ value = widget.get_value()
+ if value is not None and value != "":
+ interface_config[widget.config_key] = value
+
+ for field_key, field in self.common_fields.items():
+ widget = field["widget"]
+ value = widget.get_value()
+ if value is not None and value != "":
+ interface_config[widget.config_key] = value
+
+ # Add interface to RNS config
+ try:
+ interfaces = self.parent.app.rns.config['interfaces']
+
+ interfaces[name] = interface_config
+
+ self.parent.app.rns.config.write()
+ print(self.parent.glyphset)
+ new_item = SelectableInterfaceItem(
+ parent=self.parent,
+ name=name,
+ is_connected=False, # will always be false until restart
+ is_enabled=True,
+ iface_type=self.iface_type,
+ tx=0,
+ rx=0,
+ icon=_get_interface_icon(self.parent.glyphset, self.iface_type),
+ iface_options=interface_config
+ )
+
+ self.parent.interface_items.append(new_item)
+ self.parent._rebuild_list()
+
+ self.show_message(f"Interface {name} added. Restart NomadNet to start using this interface")
+
+ except Exception as e:
+ print(f"Error saving interface: {str(e)}")
+ self.show_message(f"Error: {str(e)}", title="Error")
+
+ def on_back(self, button):
+ self.parent.switch_to_list()
+
+ def show_message(self, message, title="Notice"):
+
+ def dismiss_dialog(button):
+ self.parent.switch_to_list()
+
+ dialog = DialogLineBox(
+ urwid.Pile([
+ urwid.Text(message, align="center"),
+ urwid.Divider(),
+ urwid.Button("OK", on_press=dismiss_dialog)
+ ]),
+ title=title
+ )
+
+ overlay = urwid.Overlay(
+ dialog,
+ self.parent.interfaces_display,
+ align='center',
+ width=100,
+ valign='middle',
+ height=8,
+ min_width=1,
+ min_height=1
+ )
+
+ self.parent.widget = overlay
+ self.parent.app.ui.main_display.update_active_sub_display()
+
+class EditInterfaceView(AddInterfaceView):
+ def __init__(self, parent, iface_name):
+ self.parent = parent
+ self.iface_name = iface_name
+
+ self.interface_config = parent.app.rns.config['interfaces'][iface_name]
+ self.iface_type = self.interface_config.get("type", "Unknown")
+
+ super().__init__(parent, self.iface_type)
+
+ self.overlay.top_w.title_widget.set_text(f"Edit Interface: {iface_name}")
+ self._populate_form_fields()
+
+ def _populate_form_fields(self):
+
+ self.fields['name']['widget'].edit_text = self.iface_name
+
+ for key, field in self.fields.items():
+ if key != 'name' and key in self.interface_config:
+ widget = field['widget']
+ value = self.interface_config[key]
+
+ if key == 'frequency':
+ # convert Hz back to MHz
+ value = float(value) / 1000000
+ # decimal format
+ value = f"{value:.6f}".rstrip('0').rstrip('.') if '.' in f"{value:.6f}" else f"{value}"
+
+ if hasattr(widget, 'edit_text'):
+ # For text input fields, set edit_text
+ widget.edit_text = str(value)
+ elif hasattr(widget, 'set_state'):
+ # For checkboxes
+ widget.set_state(bool(value))
+ elif isinstance(widget, FormDropdown):
+ # For dropdowns - update selected and update display
+ str_value = str(value)
+ if str_value in widget.options:
+ widget.selected = str_value
+ widget.main_button.base_widget.set_text(str_value)
+ else:
+ for opt in widget.options:
+ try:
+ if widget.transform(opt) == value:
+ widget.selected = opt
+ widget.main_button.base_widget.set_text(opt)
+ break
+ except:
+ pass
+ elif isinstance(widget, FormMultiList):
+ if isinstance(value, str):
+ items = [item.strip() for item in value.split(',') if item.strip()]
+ self._populate_multilist(widget, items)
+ elif isinstance(value, list):
+ self._populate_multilist(widget, value)
+
+ for key, field in self.additional_fields.items():
+ if key in self.interface_config:
+ widget = field['widget']
+ value = self.interface_config[key]
+
+ if hasattr(widget, 'edit_text'):
+ widget.edit_text = str(value)
+ elif hasattr(widget, 'set_state'):
+ # RNS.log(f"KEY: {key} VAL: {value}")
+ checkbox_state = value if isinstance(value, bool) else value.strip().lower() not in ('false', 'off', 'no', '0')
+ widget.set_state(checkbox_state)
+ elif isinstance(widget, FormDropdown):
+ str_value = str(value)
+ if str_value in widget.options:
+ widget.selected = str_value
+ widget.main_button.base_widget.set_text(str_value)
+ else:
+ # Try to match after transform
+ for opt in widget.options:
+ try:
+ if widget.transform(opt) == value:
+ widget.selected = opt
+ widget.main_button.base_widget.set_text(opt)
+ break
+ except:
+ pass
+ elif isinstance(widget, FormMultiList):
+ if isinstance(value, str):
+ items = [item.strip() for item in value.split(',') if item.strip()]
+ self._populate_multilist(widget, items)
+ elif isinstance(value, list):
+ self._populate_multilist(widget, value)
+
+ for key, field in self.common_fields.items():
+ if key in self.interface_config:
+ widget = field['widget']
+ value = self.interface_config[key]
+
+ if hasattr(widget, 'edit_text'):
+ widget.edit_text = str(value)
+ elif hasattr(widget, 'set_state'):
+ widget.set_state(bool(value))
+ elif isinstance(widget, FormDropdown):
+ str_value = str(value)
+ if str_value in widget.options:
+ widget.selected = str_value
+ widget.main_button.base_widget.set_text(str_value)
+ else:
+ # Try to match after transform
+ for opt in widget.options:
+ try:
+ if widget.transform(opt) == value:
+ widget.selected = opt
+ widget.main_button.base_widget.set_text(opt)
+ break
+ except:
+ pass
+ elif isinstance(widget, FormMultiList):
+ if isinstance(value, str):
+ items = [item.strip() for item in value.split(',') if item.strip()]
+ self._populate_multilist(widget, items)
+ elif isinstance(value, list):
+ self._populate_multilist(widget, value)
+
+ def _populate_multilist(self, widget, items):
+ while len(widget.entries) > 1:
+ widget.remove_entry(None, widget.entries[-1])
+
+ if items:
+ first_entry = widget.entries[0]
+ first_edit = first_entry.contents[0][0]
+ if len(items) > 0:
+ first_edit.edit_text = items[0]
+
+ for i in range(1, len(items)):
+ widget.add_entry(None)
+ entry = widget.entries[i]
+ edit_widget = entry.contents[0][0]
+ edit_widget.edit_text = items[i]
+
+ def on_save(self, button):
+ if not self.validate_all():
+ return
+
+ new_name = self.fields['name']["widget"].get_value() or self.iface_name
+
+ updated_config = {
+ "type": self.iface_type,
+ "interface_enabled": True
+ }
+
+ for field_key, field in self.fields.items():
+ if field_key != "name":
+ widget = field["widget"]
+ value = widget.get_value()
+ if value is not None and value != "":
+ updated_config[widget.config_key] = value
+
+ for field_key, field in self.additional_fields.items():
+ widget = field["widget"]
+ value = widget.get_value()
+ if value is not None and value != "":
+ updated_config[widget.config_key] = value
+
+ for field_key, field in self.common_fields.items():
+ widget = field["widget"]
+ value = widget.get_value()
+ if value is not None and value != "":
+ updated_config[widget.config_key] = value
+
+ try:
+ interfaces = self.parent.app.rns.config['interfaces']
+
+ if new_name != self.iface_name:
+ del interfaces[self.iface_name]
+ interfaces[new_name] = updated_config
+
+ for i, item in enumerate(self.parent.interface_items):
+ if item.name == self.iface_name:
+ self.parent.interface_items[i].name = new_name
+ break
+ else:
+ interfaces[self.iface_name] = updated_config
+
+ self.parent.app.rns.config.write()
+
+ for item in self.parent.interface_items:
+ if item.name == new_name:
+ item.iface_type = self.iface_type
+ break
+
+ self.parent._rebuild_list()
+
+ self.show_message(f"Interface {new_name} updated. Restart NomadNet for these changes to take effect")
+
+ except Exception as e:
+ print(f"Error saving interface: {str(e)}")
+ self.show_message(f"Error updating interface: {str(e)}", title="Error")
+
+
+class ShowInterface(urwid.WidgetWrap):
+ def __init__(self, parent, iface_name):
+ self.parent = parent
+ self.iface_name = iface_name
+ self.started = False
+ self.g = self.parent.app.ui.glyphs
+
+ # get config
+ self.interface_config = self.parent.app.rns.config['interfaces'][iface_name]
+ iface_type = self.interface_config.get("type", "Unknown")
+
+ self.parent.shortcuts_display.set_show_interface_shortcuts()
+
+ self.config_rows = []
+
+ # get interface stats
+ interface_stats = self.parent.app.rns.get_interface_stats()
+ stats_lookup = {iface['short_name']: iface for iface in interface_stats['interfaces']}
+ self.stats = stats_lookup.get(iface_name, {})
+
+ self.tx = self.stats.get("txb", 0)
+ self.rx = self.stats.get("rxb", 0)
+ self.is_connected = self.stats.get("status", False)
+ self.is_enabled = (str(self.interface_config.get("enabled")).lower() != 'false' and
+ str(self.interface_config.get("interface_enabled")).lower() != 'false')
+
+ header_content = [
+ urwid.Text(("interface_title", f"Interface: {iface_name}"), align="center"),
+ urwid.Divider("=")
+ ]
+ header = urwid.Pile(header_content)
+
+ self.edit_button = urwid.Button("Edit", on_press=self.on_edit)
+ self.back_button = urwid.Button("Back", on_press=self.on_back)
+
+ self.toggle_button = urwid.Button(
+ "Disable" if self.is_enabled else "Enable",
+ on_press=self.on_toggle_enabled
+ )
+
+ button_row = urwid.Columns([
+ (urwid.WEIGHT, 0.3, self.back_button),
+ (urwid.WEIGHT, 0.05, urwid.Text("")),
+ (urwid.WEIGHT, 0.3, self.toggle_button),
+ (urwid.WEIGHT, 0.05, urwid.Text("")),
+ (urwid.WEIGHT, 0.3, self.edit_button),
+ ])
+
+ footer_content = [
+ urwid.Divider("="),
+ button_row
+ ]
+ footer = urwid.Pile(footer_content)
+
+ # status widgets
+ self.status_text = urwid.Text(("connected_status" if self.is_enabled else "disconnected_status",
+ "Enabled" if self.is_enabled else "Disabled"))
+
+ self.status_indicator = urwid.Text(("connected_status" if self.is_enabled else "disconnected_status",
+ self.parent.g['selected'] if self.is_enabled else self.parent.g[
+ 'unselected']))
+
+ self.connection_text = urwid.Text(("connected_status" if self.is_connected else "disconnected_status",
+ "Connected" if self.is_connected else "Disconnected"))
+
+ self.info_rows = [
+ urwid.Columns([
+ (10, urwid.Text(("key", "Type:"))),
+ urwid.Text(("value", f"{_get_interface_icon(self.parent.glyphset, iface_type)} {iface_type}")),
+ ]),
+ urwid.Columns([
+ (10, urwid.Text(("key", "Status:"))),
+ (4, self.status_indicator),
+ (8, self.status_text),
+ (3, urwid.Text(" | ")),
+ self.connection_text,
+ ]),
+ urwid.Divider("-")
+ ]
+
+ self.tx_text = urwid.Text(("value", format_bytes(self.tx)))
+ self.rx_text = urwid.Text(("value", format_bytes(self.rx)))
+
+ self.stat_row = urwid.Columns([
+ (10, urwid.Text(("key", "TX:"))),
+ (15, self.tx_text),
+ (10, urwid.Text(("key", "RX:"))),
+ self.rx_text,
+ ])
+
+ self.info_rows.append(self.stat_row)
+ self.info_rows.append(urwid.Divider("-"))
+
+ self.bandwidth_chart = InterfaceBandwidthChart(history_length=60, glyphset=self.parent.glyphset)
+ self.bandwidth_chart.update(self.rx, self.tx)
+
+ self.rx_chart_text = urwid.Text("Loading RX data...", align='left')
+ self.tx_chart_text = urwid.Text("Loading TX data...", align='left')
+ self.rx_peak_text = urwid.Text("Peak: 0 B/s", align='right')
+ self.tx_peak_text = urwid.Text("Peak: 0 B/s", align='right')
+
+ self.rx_box = urwid.LineBox(
+ urwid.Pile([
+ urwid.AttrMap(self.rx_chart_text, "rx"),
+ self.rx_peak_text
+ ]),
+ title="RX Traffic (60s)"
+ )
+
+ self.tx_box = urwid.LineBox(
+ urwid.Pile([
+ urwid.AttrMap(self.tx_chart_text, "tx"),
+ self.tx_peak_text
+ ]),
+ title="TX Traffic (60s)"
+ )
+
+ self.horizontal_charts = urwid.Columns([
+ (urwid.WEIGHT, 1, self.rx_box),
+ (urwid.WEIGHT, 1, self.tx_box)
+ ])
+
+ self.vertical_charts = urwid.Pile([
+ self.rx_box,
+ self.tx_box
+ ])
+
+ self.disconnected_message = urwid.Filler(
+ urwid.Text(("disconnected_status",
+ "Charts not available - Interface is not connected"),
+ align="center"),
+ valign="top"
+ )
+ self.disconnected_box = urwid.LineBox(self.disconnected_message, title="Bandwidth Charts")
+
+ self.charts_widget = self.vertical_charts
+ self.is_horizontal = False
+
+ screen = urwid.raw_display.Screen()
+ screen_cols, _ = screen.get_cols_rows()
+ if screen_cols >= 100:
+ self.charts_widget = self.horizontal_charts
+ self.is_horizontal = True
+
+ if not self.is_connected:
+ self.charts_widget = self.disconnected_box
+
+ connection_params = []
+ radio_params = []
+ network_params = []
+ ifac_params = []
+ other_params = []
+
+ # Sort parameters into groups
+ for key, value in self.interface_config.items():
+ # skip empty
+ if value is None or value == "":
+ continue
+
+ # Skip these keys as their shown elsewhere
+ if key in ["type", "interface_enabled", "enabled", 'selected_interface_mode', 'name']:
+ continue
+
+ # Connection parameters
+ elif key in ["port", "listen_ip", "listen_port", "target_host", "target_port", "device"]:
+ connection_params.append((key, value))
+
+ # Radio parameters
+ elif key in ["frequency", "bandwidth", "spreadingfactor", "codingrate", "txpower"]:
+ radio_params.append((key, value))
+
+ # Network parameters
+ elif key in ["network_name", "bitrate", "peers", "group_id", "multicast_address_type",
+ "discovery_scope", "announce_cap", "mode"]:
+ network_params.append((key, value))
+
+ # IFAC parameters
+ elif key in ["passphrase", "ifac_size", "ifac_netname", "ifac_netkey"]:
+ ifac_params.append((key, value))
+
+ else:
+ other_params.append((key, value))
+
+ def create_param_row(key, value):
+ if isinstance(value, bool):
+ value_str = "Yes" if value else "No"
+ elif key == "frequency":
+ int_value = int(value)
+ value_str = f"{int_value / 1000000:.3f} MHz"
+ elif key == "bandwidth":
+ int_value = int(value)
+ value_str = f"{int_value / 1000:.1f} kHz"
+ else:
+ value_str = str(value)
+ # format display keys: "listen_port" => Listen Port
+ display_key = key.replace('_', ' ').title()
+
+ return urwid.Columns([
+ (18, urwid.Text(("key", f"{display_key}:"))),
+ urwid.Text(("value", value_str)),
+ ])
+
+ if connection_params:
+ connection_params.sort(key=lambda x: x[0])
+ self.config_rows.append(urwid.Text(("interface_title", "Connection Parameters"), align="left"))
+ for key, value in connection_params:
+ self.config_rows.append(create_param_row(key, value))
+ self.config_rows.append(urwid.Divider("-"))
+
+ if radio_params:
+ radio_params.sort(key=lambda x: x[0])
+ self.config_rows.append(urwid.Text(("interface_title", "Radio Parameters"), align="left"))
+ for key, value in radio_params:
+ self.config_rows.append(create_param_row(key, value))
+ self.config_rows.append(urwid.Divider("-"))
+
+ if network_params:
+ network_params.sort(key=lambda x: x[0])
+ self.config_rows.append(urwid.Text(("interface_title", "Network Parameters"), align="left"))
+ for key, value in network_params:
+ self.config_rows.append(create_param_row(key, value))
+ self.config_rows.append(urwid.Divider("-"))
+
+ if ifac_params:
+ ifac_params.sort(key=lambda x: x[0])
+ self.config_rows.append(urwid.Text(("interface_title", "IFAC Parameters"), align="left"))
+ for key, value in ifac_params:
+ self.config_rows.append(create_param_row(key, value))
+ self.config_rows.append(urwid.Divider("-"))
+
+ if other_params:
+ other_params.sort(key=lambda x: x[0])
+ self.config_rows.append(urwid.Text(("interface_title", "Additional Parameters"), align="left"))
+ for key, value in other_params:
+ self.config_rows.append(create_param_row(key, value))
+ self.config_rows.append(urwid.Divider("-"))
+
+ if not self.config_rows:
+ self.config_rows.append(urwid.Text("No additional parameters", align="center"))
+
+ body_content = []
+ body_content.extend(self.info_rows)
+ body_content.append(
+ self.charts_widget)
+ body_content.append(urwid.Divider("-"))
+ body_content.extend(self.config_rows)
+
+ body_pile = urwid.Pile(body_content)
+ body_padding = urwid.Padding(body_pile, left=2, right=2)
+
+ body = urwid.ListBox(urwid.SimpleListWalker([body_padding]))
+
+ self.frame = urwid.Frame(
+ body=body,
+ header=header,
+ footer=footer
+ )
+
+ self.content_box = urwid.LineBox(self.frame)
+
+ super().__init__(self.content_box)
+
+ def update_status_display(self):
+ if self.is_enabled:
+ self.status_indicator.set_text(("connected_status", self.parent.g['selected']))
+ self.status_text.set_text(("connected_status", "Enabled"))
+ else:
+ self.status_indicator.set_text(("disconnected_status", self.parent.g['unselected']))
+ self.status_text.set_text(("disconnected_status", "Disabled"))
+
+ def update_connection_display(self, is_connected):
+ old_connection_state = self.is_connected
+ self.is_connected = is_connected
+
+ self.connection_text.set_text(("connected_status" if self.is_connected else "disconnected_status",
+ "Connected" if self.is_connected else "Disconnected"))
+
+ if old_connection_state != self.is_connected:
+ body_pile = self.frame.body.body[0].original_widget
+
+ chart_index = None
+ for i, (widget, options) in enumerate(body_pile.contents):
+ if (widget == self.horizontal_charts or
+ widget == self.vertical_charts or
+ widget == self.disconnected_box):
+ chart_index = i
+ break
+
+ if chart_index is not None:
+ if self.is_connected:
+ new_widget = self.horizontal_charts if self.is_horizontal else self.vertical_charts
+ if not self.started:
+ self.start()
+ else:
+ new_widget = self.disconnected_box
+ self.started = False
+
+ body_pile.contents[chart_index] = (new_widget, body_pile.options())
+
+ def on_toggle_enabled(self, button):
+ action = "disable" if self.is_enabled else "enable"
+
+ def on_confirm_yes(confirm_button):
+ self.parent.app.ui.main_display.frame.body = self.parent.app.ui.main_display.sub_displays.active().widget
+
+ self.is_enabled = not self.is_enabled
+
+ self.toggle_button.set_label("Disable" if self.is_enabled else "Enable")
+
+ if "interface_enabled" in self.interface_config:
+ self.interface_config["interface_enabled"] = self.is_enabled
+ else:
+ self.interface_config["enabled"] = self.is_enabled
+
+ try:
+ interfaces = self.parent.app.rns.config['interfaces']
+
+ interfaces[self.iface_name] = self.interface_config
+
+ self.parent.app.rns.config.write()
+
+ self.update_status_display()
+
+ for item in self.parent.interface_items:
+ if item.name == self.iface_name:
+ item.is_enabled = self.is_enabled
+ item.update_status_display()
+
+ if hasattr(self.parent.app.ui, 'loop') and self.parent.app.ui.loop is not None:
+ self.parent.app.ui.loop.draw_screen()
+
+ self.show_restart_required_message()
+
+ except Exception as e:
+ self.show_error_message(f"Error updating interface: {str(e)}")
+
+ def on_confirm_no(confirm_button):
+ self.parent.app.ui.main_display.frame.body = self.parent.app.ui.main_display.sub_displays.active().widget
+
+ confirm_text = urwid.Text((
+ "interface_title",
+ f"Are you sure you want to {action} the {self.iface_name} interface?"
+ ), align="center")
+
+ yes_button = urwid.Button("Yes", on_press=on_confirm_yes)
+ no_button = urwid.Button("No", on_press=on_confirm_no)
+
+ buttons_row = urwid.Columns([
+ (urwid.WEIGHT, 0.45, yes_button),
+ (urwid.WEIGHT, 0.1, urwid.Text("")),
+ (urwid.WEIGHT, 0.45, no_button),
+ ])
+
+ pile = urwid.Pile([
+ confirm_text,
+ urwid.Divider(),
+ buttons_row
+ ])
+
+ dialog = DialogLineBox(pile, title="Confirm")
+
+ overlay = urwid.Overlay(
+ dialog,
+ self.parent.app.ui.main_display.frame.body,
+ align='center',
+ width=50,
+ valign='middle',
+ height=7
+ )
+
+ self.parent.app.ui.main_display.frame.body = overlay
+
+ def show_restart_required_message(self):
+
+ def dismiss_dialog(button):
+ self.parent.app.ui.main_display.frame.body = self.parent.app.ui.main_display.sub_displays.active().widget
+
+ dialog = DialogLineBox(
+ urwid.Pile([
+ urwid.Text(
+ f"Interface {self.iface_name} has been " +
+ ("enabled" if self.is_enabled else "disabled") +
+ ".\nRestart required for changes to take effect.",
+ align="center"
+ ),
+ urwid.Divider(),
+ urwid.Button("OK", on_press=dismiss_dialog)
+ ]),
+ title="Notice"
+ )
+
+ overlay = urwid.Overlay(
+ dialog,
+ self.parent.app.ui.main_display.frame.body,
+ align='center',
+ width=50,
+ valign='middle',
+ height=8
+ )
+
+ self.parent.app.ui.main_display.frame.body = overlay
+
+ def show_error_message(self, message):
+
+ def dismiss_dialog(button):
+ self.parent.app.ui.main_display.frame.body = self.parent.app.ui.main_display.sub_displays.active().widget
+
+ dialog = DialogLineBox(
+ urwid.Pile([
+ urwid.Text(message, align="center"),
+ urwid.Divider(),
+ urwid.Button("OK", on_press=dismiss_dialog)
+ ]),
+ title="Error"
+ )
+
+ overlay = urwid.Overlay(
+ dialog,
+ self.parent.app.ui.main_display.frame.body,
+ align='center',
+ width=50,
+ valign='middle',
+ height=8
+ )
+
+ self.parent.app.ui.main_display.frame.body = overlay
+
+ def keypress(self, size, key):
+ if key == 'tab':
+ if self.frame.focus_position == 'body':
+ self.frame.focus_position = 'footer'
+ footer_pile = self.frame.footer
+ if isinstance(footer_pile, urwid.Pile):
+ footer_pile.focus_position = 1 # button row
+ button_row = footer_pile.contents[1][0]
+ if isinstance(button_row, urwid.Columns):
+ button_row.focus_position = 0
+ return None
+ # If we're currently in footer change the focus
+ elif self.frame.focus_position == 'footer':
+ footer_pile = self.frame.footer
+ if isinstance(footer_pile, urwid.Pile):
+ button_row = footer_pile.contents[1][0]
+ if isinstance(button_row, urwid.Columns):
+ # If on first button (Back), move to Toggle button
+ if button_row.focus_position == 0:
+ button_row.focus_position = 2 # skip spacer
+ return None
+ # If on toggle button, move to Edit button
+ elif button_row.focus_position == 2:
+ button_row.focus_position = 4
+ return None
+ # if on edit button wrap back to toggle button
+ elif button_row.focus_position == 4:
+ button_row.focus_position = 0
+ return None
+ elif key == 'shift tab':
+ if self.frame.focus_position == 'footer':
+ self.frame.focus_position = 'body'
+ return None
+ elif self.frame.focus_position == 'footer':
+ footer_pile = self.frame.footer
+ if isinstance(footer_pile, urwid.Pile):
+ button_row = footer_pile.contents[1][0]
+ if isinstance(button_row, urwid.Columns):
+ if button_row.focus_position == 4: # edit button
+ button_row.focus_position = 2 # toggle button
+ return None
+ elif button_row.focus_position == 2:
+ button_row.focus_position = 0 # back button
+ return None
+ elif button_row.focus_position == 0: # back button
+ self.frame.focus_position = 'body'
+ return None
+ elif key == "h" and self.is_connected: # horizontal layout
+ if not self.is_horizontal:
+ self.switch_to_horizontal()
+ return None
+ elif key == "v" and self.is_connected: # vertical layout
+ if self.is_horizontal:
+ self.switch_to_vertical()
+ return None
+
+ return super().keypress(size, key)
+
+ def switch_to_horizontal(self):
+ if not self.is_connected:
+ return
+
+ self.is_horizontal = True
+
+ body_pile = self.frame.body.body[0].original_widget
+ for i, (widget, options) in enumerate(body_pile.contents):
+ if widget == self.vertical_charts:
+ body_pile.contents[i] = (self.horizontal_charts, options)
+ self.charts_widget = self.horizontal_charts
+ break
+
+ def switch_to_vertical(self):
+ if not self.is_connected:
+ return
+
+ self.is_horizontal = False
+ body_pile = self.frame.body.body[0].original_widget
+ for i, (widget, options) in enumerate(body_pile.contents):
+ if widget == self.horizontal_charts:
+ body_pile.contents[i] = (self.vertical_charts, options)
+ self.charts_widget = self.vertical_charts
+ break
+
+ def start(self):
+ if not self.started and self.is_connected:
+ self.started = True
+ self.parent.app.ui.loop.set_alarm_in(1, self.update_bandwidth_charts)
+
+ def update_bandwidth_charts(self, loop, user_data):
+ if not self.started:
+ return
+
+ interface_stats = self.parent.app.rns.get_interface_stats()
+ stats_lookup = {iface['short_name']: iface for iface in interface_stats['interfaces']}
+ stats = stats_lookup.get(self.iface_name, {})
+
+ tx = stats.get("txb", self.tx)
+ rx = stats.get("rxb", self.rx)
+
+ new_connection_status = stats.get("status", False)
+ if new_connection_status != self.is_connected:
+ self.update_connection_display(new_connection_status)
+
+ if not self.is_connected:
+ return
+
+ self.tx_text.set_text(("value", format_bytes(tx)))
+ self.rx_text.set_text(("value", format_bytes(rx)))
+
+ self.bandwidth_chart.update(rx, tx)
+
+ rx_chart, tx_chart, peak_rx, peak_tx = self.bandwidth_chart.get_charts(height=8)
+
+ self.rx_chart_text.set_text(rx_chart)
+ self.tx_chart_text.set_text(tx_chart)
+ self.rx_peak_text.set_text(f"Peak: {peak_rx}")
+ self.tx_peak_text.set_text(f"Peak: {peak_tx}")
+
+ self.tx = tx
+ self.rx = rx
+
+ if self.started:
+ loop.set_alarm_in(1, self.update_bandwidth_charts)
+
+ def on_back(self, button):
+ self.started = False
+ self.parent.switch_to_list()
+
+ def on_edit(self, button):
+ self.started = False
+ self.parent.switch_to_edit_interface(self.iface_name)
+
+### MAIN DISPLAY ###
+class InterfaceDisplay:
+ def __init__(self, app):
+ self.app = app
+ self.started = False
+ self.interface_items = []
+ self.glyphset = self.app.config["textui"]["glyphs"]
+ self.g = self.app.ui.glyphs
+
+ self.terminal_cols, self.terminal_rows = urwid.raw_display.Screen().get_cols_rows()
+ self.iface_row_offset = 4
+ self.list_rows = self.terminal_rows - self.iface_row_offset
+
+ interfaces = app.rns.config['interfaces']
+ processed_interfaces = {}
+
+ for interface_name, interface in interfaces.items():
+ interface_data = interface.copy()
+
+ # handle sub-interfaces for RNodeMultiInterface
+ if interface_data.get("type") == "RNodeMultiInterface":
+ sub_interfaces = []
+ for sub_name, sub_config in interface_data.items():
+ if sub_name not in {"type", "port", "interface_enabled", "selected_interface_mode",
+ "configured_bitrate"}:
+ if isinstance(sub_config, dict):
+ sub_config["name"] = sub_name
+ sub_interfaces.append(sub_config)
+
+ # add sub-interfaces to the main interface data
+ interface_data["sub_interfaces"] = sub_interfaces
+
+ for sub in sub_interfaces:
+ del interface_data[sub["name"]]
+
+ processed_interfaces[interface_name] = interface_data
+
+ interface_stats = app.rns.get_interface_stats()
+ stats_lookup = {interface['short_name']: interface for interface in interface_stats['interfaces']}
+ # print(stats_lookup)
+ for interface_name, interface_data in processed_interfaces.items():
+ # configobj false values
+ is_enabled = str(interface_data.get("enabled")).lower() not in ('false', 'off', 'no', '0') and str(interface_data.get("interface_enabled")).lower() not in ('false', 'off', 'no', '0')
+
+ iface_type = interface_data.get("type", "Unknown")
+ icon = _get_interface_icon(self.glyphset, iface_type)
+
+ stats_for_interface = stats_lookup.get(interface_name)
+
+ if stats_for_interface:
+ tx = stats_for_interface.get("txb", 0)
+ rx = stats_for_interface.get("rxb", 0)
+ is_connected = stats_for_interface["status"]
+ else:
+ tx = 0
+ rx = 0
+ is_connected = False
+
+ item = SelectableInterfaceItem(
+ parent=self,
+ name=interface_data.get("name", interface_name),
+ is_connected=is_connected,
+ is_enabled=is_enabled,
+ iface_type=iface_type,
+ tx=tx,
+ rx=rx,
+ icon=icon
+ )
+
+ self.interface_items.append(item)
+
+ interface_header = urwid.Text(("interface_title", "Interfaces"), align="center")
+ if len(self.interface_items) == 0:
+ interface_header = urwid.Text(
+ ("interface_title", "No interfaces found. Press Ctrl + A to add a new interface "), align="center")
+
+
+ list_contents = [
+ interface_header,
+ urwid.Divider(),
+ ] + self.interface_items
+
+ self.list_walker = urwid.SimpleFocusListWalker(list_contents)
+ self.list_box = urwid.ListBox(self.list_walker)
+
+ self.box_adapter = urwid.BoxAdapter(self.list_box, self.list_rows)
+
+
+ pile = urwid.Pile([self.box_adapter])
+ self.interfaces_display = InterfaceFiller(pile, self.app)
+ self.shortcuts_display = InterfaceDisplayShortcuts(self.app)
+ self.widget = self.interfaces_display
+
+ def start(self):
+ # started from Main.py
+ self.started = True
+ self.app.ui.loop.set_alarm_in(1, self.poll_stats)
+ self.app.ui.loop.set_alarm_in(5, self.check_terminal_size)
+
+ def switch_to_edit_interface(self, iface_name):
+ self.edit_interface_view = EditInterfaceView(self, iface_name)
+ self.widget = self.edit_interface_view
+ self.app.ui.main_display.update_active_sub_display()
+
+ def edit_selected_interface(self):
+ focus_widget, focus_position = self.box_adapter._original_widget.body.get_focus()
+
+ if not isinstance(focus_widget, SelectableInterfaceItem):
+ return
+
+ selected_item = focus_widget
+ interface_name = selected_item.name
+
+ self.switch_to_edit_interface(interface_name)
+
+ def check_terminal_size(self, loop, user_data):
+ new_cols, new_rows = loop.screen.get_cols_rows()
+
+ if new_rows != self.terminal_rows or new_cols != self.terminal_cols:
+ self.terminal_cols, self.terminal_rows = new_cols, new_rows
+ self.list_rows = self.terminal_rows - self.iface_row_offset
+
+ self.box_adapter.height = self.list_rows
+
+ loop.draw_screen()
+
+ if self.started:
+ loop.set_alarm_in(5, self.check_terminal_size)
+
+ def poll_stats(self, loop, user_data):
+ interface_stats = self.app.rns.get_interface_stats()
+ stats_lookup = {iface['short_name']: iface for iface in interface_stats['interfaces']}
+ for item in self.interface_items:
+ # use interface name as the key
+ stats_for_interface = stats_lookup.get(item.name)
+ if stats_for_interface:
+ tx = stats_for_interface.get("txb", 0)
+ rx = stats_for_interface.get("rxb", 0)
+ item.update_stats(tx, rx)
+
+ loop.set_alarm_in(1, self.poll_stats)
+
+
+ def shortcuts(self):
+ return self.shortcuts_display
+
+ def switch_to_show_interface(self, iface_name):
+ show_interface = ShowInterface(self, iface_name)
+ self.widget = show_interface
+ self.app.ui.main_display.update_active_sub_display()
+
+ show_interface.start()
+
+ def switch_to_list(self):
+ self.shortcuts_display.reset_shortcuts()
+ self.widget = self.interfaces_display
+ self._rebuild_list()
+ self.app.ui.main_display.update_active_sub_display()
+
+ def add_interface(self):
+ dialog_widgets = []
+
+ def add_heading(txt):
+ dialog_widgets.append(urwid.Text(("interface_title", txt), align="left"))
+
+ def add_option(label, value):
+ item = InterfaceOptionItem(self, label, value)
+ dialog_widgets.append(item)
+
+ # Get the icons based on plain, unicode, nerdfont glyphset
+ network_icon = _get_interface_icon(self.glyphset, "AutoInterface")
+ rnode_icon = _get_interface_icon(self.glyphset, "RNodeInterface")
+ serial_icon = _get_interface_icon(self.glyphset, "SerialInterface")
+ other_icon = _get_interface_icon(self.glyphset, "PipeInterface")
+
+ add_heading(f"{network_icon} IP Networks")
+ add_option("Auto Interface", "AutoInterface")
+ add_option("TCP Client Interface", "TCPClientInterface")
+ add_option("TCP Server Interface", "TCPServerInterface")
+ add_option("UDP Interface", "UDPInterface")
+ add_option("I2P Interface", "I2PInterface")
+
+ add_heading(f"{rnode_icon} RNodes")
+ add_option("RNode Interface", "RNodeInterface")
+ # add_option("RNode Multi Interface", "RNodeMultiInterface") TODO
+
+ add_heading(f"{serial_icon} Hardware")
+ add_option("Serial Interface", "SerialInterface")
+ add_option("KISS Interface", "KISSInterface")
+ add_option("AX.25 KISS Interface", "AX25KISSInterface")
+
+ add_heading(f"{other_icon} Other")
+ add_option("Pipe Interface", "PipeInterface")
+ # add_option("Custom Interface", "CustomInterface") TODO
+
+ listbox = urwid.ListBox(urwid.SimpleFocusListWalker(dialog_widgets))
+ dialog = DialogLineBox(listbox, parent=self, title="Select Interface Type")
+
+ overlay = urwid.Overlay(
+ dialog,
+ self.interfaces_display,
+ align='center',
+ width=('relative', 50),
+ valign='middle',
+ height=('relative', 50),
+ min_width=20,
+ min_height=15,
+ left=2,
+ right=2
+ )
+ self.widget = overlay
+ self.app.ui.main_display.update_active_sub_display()
+
+ def switch_to_add_interface(self, iface_type):
+ self.add_interface_view = AddInterfaceView(self, iface_type)
+ self.widget = self.add_interface_view
+ self.app.ui.main_display.update_active_sub_display()
+
+ def remove_selected_interface(self):
+ focus_widget, focus_position = self.box_adapter._original_widget.body.get_focus()
+ if not isinstance(focus_widget, SelectableInterfaceItem):
+ return
+
+ selected_item = focus_widget
+ interface_name = selected_item.name
+
+ def on_confirm_yes(button):
+ try:
+ if interface_name in self.app.rns.config['interfaces']:
+ del self.app.rns.config['interfaces'][interface_name]
+ self.app.rns.config.write()
+
+ if selected_item in self.interface_items:
+ self.interface_items.remove(selected_item)
+
+ self._rebuild_list()
+ self.dismiss_dialog()
+
+ except Exception as e:
+ print(e)
+
+ def on_confirm_no(button):
+ self.dismiss_dialog()
+
+ confirm_text = urwid.Text(("interface_title", f"Remove interface {interface_name}?"), align="center")
+ yes_button = urwid.Button("Yes", on_press=on_confirm_yes)
+ no_button = urwid.Button("No", on_press=on_confirm_no)
+
+ buttons_row = urwid.Columns([
+ (urwid.WEIGHT, 0.45, yes_button),
+ (urwid.WEIGHT, 0.1, urwid.Text("")),
+ (urwid.WEIGHT, 0.45, no_button),
+ ])
+
+ pile = urwid.Pile([
+ confirm_text,
+ buttons_row
+ ])
+
+ dialog = DialogLineBox(pile, parent=self, title="?")
+
+ overlay = urwid.Overlay(
+ dialog,
+ self.interfaces_display,
+ align='center',
+ width=('relative', 35),
+ valign='middle',
+ height=(5),
+ min_width=5,
+ left=2,
+ right=2
+ )
+ dialog.original_widget.focus_position = 1 # columns row
+ buttons_row = dialog.original_widget.contents[1][0]
+ buttons_row.focus_position = 2 # second button "No"
+
+ self.widget = overlay
+ self.app.ui.main_display.update_active_sub_display()
+
+ def dismiss_dialog(self):
+ self.widget = self.interfaces_display
+ self.app.ui.main_display.update_active_sub_display()
+
+ def _rebuild_list(self):
+ interface_header = urwid.Text(("interface_title", f"Interfaces ({len(self.interface_items)})"), align="center")
+ if len(self.interface_items) == 0:
+ interface_header = urwid.Text(("interface_title", "No interfaces found. Press Ctrl + A to add a new interface "), align="center")
+
+ new_list = [
+ interface_header,
+ urwid.Divider(),
+ ] + self.interface_items
+ # RNS.log(f"items: {self.interface_items}")
+
+ walker = urwid.SimpleFocusListWalker(new_list)
+ self.box_adapter._original_widget.body = walker
+ self.box_adapter._original_widget.focus_position = len(new_list) - 1
+
+### SHORTCUTS ###
+class InterfaceDisplayShortcuts:
+ def __init__(self, app):
+ self.app = app
+ self.default_shortcuts = "[C-a] Add Interface [C-e] Edit Interface [C-x] Remove Interface [Enter] Show Interface"
+ self.current_shortcuts = self.default_shortcuts
+ self.widget = urwid.AttrMap(
+ urwid.Text(self.current_shortcuts),
+ "shortcutbar"
+ )
+
+ def update_shortcuts(self, new_shortcuts):
+ self.current_shortcuts = new_shortcuts
+ self.widget.original_widget.set_text(new_shortcuts)
+
+ def reset_shortcuts(self):
+ self.update_shortcuts(self.default_shortcuts)
+
+ def set_show_interface_shortcuts(self):
+ show_shortcuts = "[Back] Return to List [Tab] Navigate [Shift-tab] Change Focus [h] Horizontal Charts [v] Vertical Charts [Up/Down] Scroll"
+ self.update_shortcuts(show_shortcuts)
+
+ def set_add_interface_shortcuts(self):
+ add_shortcuts = "[Up/Down] Navigate Fields [Enter] Select Option"
+ self.update_shortcuts(add_shortcuts)
+
+ def set_edit_interface_shortcuts(self):
+ edit_shortcuts = "[Up/Down] Navigate Fields [Enter] Select Option"
+ self.update_shortcuts(edit_shortcuts)
\ No newline at end of file
diff --git a/nomadnet/ui/textui/Main.py b/nomadnet/ui/textui/Main.py
index 303024a..34d88c3 100644
--- a/nomadnet/ui/textui/Main.py
+++ b/nomadnet/ui/textui/Main.py
@@ -4,6 +4,7 @@ from .Network import *
from .Conversations import *
from .Directory import *
from .Config import *
+from .Interfaces import *
from .Map import *
from .Log import *
from .Guide import *
@@ -16,6 +17,7 @@ class SubDisplays():
self.conversations_display = ConversationsDisplay(self.app)
self.directory_display = DirectoryDisplay(self.app)
self.config_display = ConfigDisplay(self.app)
+ self.interface_display = InterfaceDisplay(self.app)
self.map_display = MapDisplay(self.app)
self.log_display = LogDisplay(self.app)
self.guide_display = GuideDisplay(self.app)
@@ -113,6 +115,11 @@ class MainDisplay():
self.sub_displays.active_display = self.sub_displays.config_display
self.update_active_sub_display()
+ def show_interfaces(self, user_data):
+ self.sub_displays.active_display = self.sub_displays.interface_display
+ self.update_active_sub_display()
+ self.sub_displays.interface_display.start()
+
def show_log(self, user_data):
self.sub_displays.active_display = self.sub_displays.log_display
self.sub_displays.log_display.show()
@@ -171,21 +178,22 @@ class MenuDisplay():
self.menu_indicator = urwid.Text("")
- menu_text = (urwid.PACK, self.menu_indicator)
- button_network = (11, MenuButton("Network", on_press=handler.show_network))
- button_conversations = (17, MenuButton("Conversations", on_press=handler.show_conversations))
- button_directory = (13, MenuButton("Directory", on_press=handler.show_directory))
- button_map = (7, MenuButton("Map", on_press=handler.show_map))
- button_log = (7, MenuButton("Log", on_press=handler.show_log))
- button_config = (10, MenuButton("Config", on_press=handler.show_config))
- button_guide = (9, MenuButton("Guide", on_press=handler.show_guide))
- button_quit = (8, MenuButton("Quit", on_press=handler.quit))
+ menu_text = (urwid.PACK, self.menu_indicator)
+ button_network = (11, MenuButton("Network", on_press=handler.show_network))
+ button_conversations = (17, MenuButton("Conversations", on_press=handler.show_conversations))
+ button_directory = (13, MenuButton("Directory", on_press=handler.show_directory))
+ button_map = (7, MenuButton("Map", on_press=handler.show_map))
+ button_log = (7, MenuButton("Log", on_press=handler.show_log))
+ button_config = (10, MenuButton("Config", on_press=handler.show_config))
+ button_interfaces = (14, MenuButton("Interfaces", on_press=handler.show_interfaces))
+ button_guide = (9, MenuButton("Guide", on_press=handler.show_guide))
+ button_quit = (8, MenuButton("Quit", on_press=handler.quit))
# buttons = [menu_text, button_conversations, button_node, button_directory, button_map]
if self.app.config["textui"]["hide_guide"]:
- buttons = [menu_text, button_conversations, button_network, button_log, button_config, button_quit]
+ buttons = [menu_text, button_conversations, button_network, button_log, button_interfaces, button_config, button_quit]
else:
- buttons = [menu_text, button_conversations, button_network, button_log, button_config, button_guide, button_quit]
+ buttons = [menu_text, button_conversations, button_network, button_log, button_interfaces, button_config, button_guide, button_quit]
columns = MenuColumns(buttons, dividechars=1)
columns.handler = handler
diff --git a/nomadnet/vendor/AsciiChart.py b/nomadnet/vendor/AsciiChart.py
new file mode 100644
index 0000000..4689590
--- /dev/null
+++ b/nomadnet/vendor/AsciiChart.py
@@ -0,0 +1,87 @@
+from __future__ import division
+from math import ceil, floor, isnan
+# Derived from asciichartpy | https://github.com/kroitor/asciichart/blob/master/asciichartpy/__init__.py
+class AsciiChart:
+ def __init__(self, glyphset="unicode"):
+ self.symbols = ['┼', '┤', '╶', '╴', '─', '╰', '╭', '╮', '╯', '│']
+ if glyphset == "plain":
+ self.symbols = ['+', '|', '-', '-', '-', '\'', ',', '.', '`', '|']
+ def plot(self, series, cfg=None):
+ if len(series) == 0:
+ return ''
+ if not isinstance(series[0], list):
+ if all(isnan(n) for n in series):
+ return ''
+ else:
+ series = [series]
+ cfg = cfg or {}
+ minimum = cfg.get('min', min(filter(lambda n: not isnan(n), [j for i in series for j in i])))
+ maximum = cfg.get('max', max(filter(lambda n: not isnan(n), [j for i in series for j in i])))
+ symbols = cfg.get('symbols', self.symbols)
+ if minimum > maximum:
+ raise ValueError('The min value cannot exceed the max value.')
+ interval = maximum - minimum
+ offset = cfg.get('offset', 3)
+ height = cfg.get('height', interval)
+ ratio = height / interval if interval > 0 else 1
+
+ min2 = int(floor(minimum * ratio))
+ max2 = int(ceil(maximum * ratio))
+
+ def clamp(n):
+ return min(max(n, minimum), maximum)
+
+ def scaled(y):
+ return int(round(clamp(y) * ratio) - min2)
+
+ rows = max2 - min2
+
+ width = 0
+ for i in range(0, len(series)):
+ width = max(width, len(series[i]))
+ width += offset
+
+ placeholder = cfg.get('format', '{:8.2f} ')
+
+ result = [[' '] * width for i in range(rows + 1)]
+
+ for y in range(min2, max2 + 1):
+ label = placeholder.format(maximum - ((y - min2) * interval / (rows if rows else 1)))
+ result[y - min2][max(offset - len(label), 0)] = label
+ result[y - min2][offset - 1] = symbols[0] if y == 0 else symbols[1]
+
+ d0 = series[0][0]
+ if not isnan(d0):
+ result[rows - scaled(d0)][offset - 1] = symbols[0]
+
+ for i in range(0, len(series)):
+ for x in range(0, len(series[i]) - 1):
+ d0 = series[i][x + 0]
+ d1 = series[i][x + 1]
+
+ if isnan(d0) and isnan(d1):
+ continue
+
+ if isnan(d0) and not isnan(d1):
+ result[rows - scaled(d1)][x + offset] = symbols[2]
+ continue
+
+ if not isnan(d0) and isnan(d1):
+ result[rows - scaled(d0)][x + offset] = symbols[3]
+ continue
+
+ y0 = scaled(d0)
+ y1 = scaled(d1)
+ if y0 == y1:
+ result[rows - y0][x + offset] = symbols[4]
+ continue
+
+ result[rows - y1][x + offset] = symbols[5] if y0 > y1 else symbols[6]
+ result[rows - y0][x + offset] = symbols[7] if y0 > y1 else symbols[8]
+
+ start = min(y0, y1) + 1
+ end = max(y0, y1)
+ for y in range(start, end):
+ result[rows - y][x + offset] = symbols[9]
+
+ return '\n'.join([''.join(row).rstrip() for row in result])
\ No newline at end of file
diff --git a/nomadnet/vendor/additional_urwid_widgets/FormWidgets.py b/nomadnet/vendor/additional_urwid_widgets/FormWidgets.py
new file mode 100644
index 0000000..4e4809a
--- /dev/null
+++ b/nomadnet/vendor/additional_urwid_widgets/FormWidgets.py
@@ -0,0 +1,283 @@
+import urwid
+
+class DialogLineBox(urwid.LineBox):
+ def __init__(self, body, parent=None, title="?"):
+ super().__init__(body, title=title)
+ self.parent = parent
+
+ def keypress(self, size, key):
+ if key == "esc":
+ if self.parent and hasattr(self.parent, "dismiss_dialog"):
+ self.parent.dismiss_dialog()
+ return None
+ return super().keypress(size, key)
+
+class Placeholder(urwid.Edit):
+ def __init__(self, caption="", edit_text="", placeholder="", **kwargs):
+ super().__init__(caption, edit_text, **kwargs)
+ self.placeholder = placeholder
+
+ def render(self, size, focus=False):
+ if not self.edit_text and not focus:
+ placeholder_widget = urwid.Text(("placeholder", self.placeholder))
+ return placeholder_widget.render(size, focus)
+ else:
+ return super().render(size, focus)
+
+class Dropdown(urwid.WidgetWrap):
+ signals = ['change'] # emit for urwid.connect_signal fn
+
+ def __init__(self, label, options, default=None):
+ self.label = label
+ self.options = options
+ self.selected = default if default is not None else options[0]
+
+ self.main_text = f"{self.selected}"
+ self.main_button = urwid.SelectableIcon(self.main_text, 0)
+ self.main_button = urwid.AttrMap(self.main_button, "button_normal", "button_focus")
+
+ self.option_widgets = []
+ for opt in options:
+ icon = urwid.SelectableIcon(opt, 0)
+ icon = urwid.AttrMap(icon, "list_normal", "list_focus")
+ self.option_widgets.append(icon)
+
+ self.options_walker = urwid.SimpleFocusListWalker(self.option_widgets)
+ self.options_listbox = urwid.ListBox(self.options_walker)
+ self.dropdown_box = None # will be created on open_dropdown
+
+ self.pile = urwid.Pile([self.main_button])
+ self.dropdown_visible = False
+
+ super().__init__(self.pile)
+
+ def open_dropdown(self):
+ if not self.dropdown_visible:
+ height = len(self.options)
+ self.dropdown_box = urwid.BoxAdapter(self.options_listbox, height)
+ self.pile.contents.append((self.dropdown_box, self.pile.options()))
+ self.dropdown_visible = True
+ self.pile.focus_position = 1
+ self.options_walker.set_focus(0)
+
+ def close_dropdown(self):
+ if self.dropdown_visible:
+ self.pile.contents.pop() # remove the dropdown_box
+ self.dropdown_visible = False
+ self.pile.focus_position = 0
+ self.dropdown_box = None
+
+ def keypress(self, size, key):
+ if not self.dropdown_visible:
+ if key == "enter":
+ self.open_dropdown()
+ return None
+ return self.main_button.keypress(size, key)
+ else:
+ if key == "enter":
+ focus_result = self.options_walker.get_focus()
+ if focus_result is not None:
+ focus_widget = focus_result[0]
+ new_val = focus_widget.base_widget.text
+ old_val = self.selected
+ self.selected = new_val
+ self.main_button.base_widget.set_text(f"{self.selected}")
+
+ if old_val != new_val:
+ self._emit('change', new_val)
+
+ self.close_dropdown()
+ return None
+ return self.dropdown_box.keypress(size, key)
+
+ def get_value(self):
+ return self.selected
+
+class ValidationError(urwid.Text):
+ def __init__(self, message=""):
+ super().__init__(("error", message))
+
+class FormField:
+ def __init__(self, config_key, transform=None):
+ self.config_key = config_key
+ self.transform = transform or (lambda x: x)
+
+class FormEdit(Placeholder, FormField):
+ def __init__(self, config_key, caption="", edit_text="", placeholder="", validation_types=None, transform=None, **kwargs):
+ Placeholder.__init__(self, caption, edit_text, placeholder, **kwargs)
+ FormField.__init__(self, config_key, transform)
+ self.validation_types = validation_types or []
+ self.error_widget = urwid.Text("")
+ self.error = None
+
+ def get_value(self):
+ return self.transform(self.edit_text.strip())
+
+ def validate(self):
+ value = self.edit_text.strip()
+ self.error = None
+
+ for validation in self.validation_types:
+ if validation == "required":
+ if not value:
+ self.error = "This field is required"
+ break
+ elif validation == "number":
+ if value and not value.replace('-', '').replace('.', '').isdigit():
+ self.error = "This field must be a number"
+ break
+ elif validation == "float":
+ try:
+ if value:
+ float(value)
+ except ValueError:
+ self.error = "This field must be decimal number"
+ break
+
+ self.error_widget.set_text(("error", self.error or ""))
+ return self.error is None
+
+class FormCheckbox(urwid.CheckBox, FormField):
+ def __init__(self, config_key, label="", state=False, validation_types=None, transform=None, **kwargs):
+ urwid.CheckBox.__init__(self, label, state, **kwargs)
+ FormField.__init__(self, config_key, transform)
+ self.validation_types = validation_types or []
+ self.error_widget = urwid.Text("")
+ self.error = None
+
+ def get_value(self):
+ return self.transform(self.get_state())
+
+ def validate(self):
+
+ value = self.get_state()
+ self.error = None
+
+ for validation in self.validation_types:
+ if validation == "required":
+ if not value:
+ self.error = "This field is required"
+ break
+
+ self.error_widget.set_text(("error", self.error or ""))
+ return self.error is None
+
+class FormDropdown(Dropdown, FormField):
+ signals = ['change']
+
+ def __init__(self, config_key, label, options, default=None, validation_types=None, transform=None):
+ self.options = [str(opt) for opt in options]
+
+ if default is not None:
+ default_str = str(default)
+ if default_str in self.options:
+ default = default_str
+ elif transform:
+ try:
+ default_transformed = transform(default_str)
+ for opt in self.options:
+ if transform(opt) == default_transformed:
+ default = opt
+ break
+ except:
+ default = self.options[0]
+ else:
+ default = self.options[0]
+ else:
+ default = self.options[0]
+
+ Dropdown.__init__(self, label, self.options, default)
+ FormField.__init__(self, config_key, transform)
+
+ self.validation_types = validation_types or []
+ self.error_widget = urwid.Text("")
+ self.error = None
+
+ if hasattr(self, 'main_button'):
+ self.main_button.base_widget.set_text(str(default))
+
+ def get_value(self):
+ return self.transform(self.selected)
+
+ def validate(self):
+ value = self.get_value()
+ self.error = None
+
+ for validation in self.validation_types:
+ if validation == "required":
+ if not value:
+ self.error = "This field is required"
+ break
+
+ self.error_widget.set_text(("error", self.error or ""))
+ return self.error is None
+
+ def open_dropdown(self):
+ if not self.dropdown_visible:
+ super().open_dropdown()
+ try:
+ current_index = self.options.index(self.selected)
+ self.options_walker.set_focus(current_index)
+ except ValueError:
+ pass
+
+class FormMultiList(urwid.Pile, FormField):
+ def __init__(self, config_key, placeholder="", validation_types=None, transform=None, **kwargs):
+ self.entries = []
+ self.error_widget = urwid.Text("")
+ self.error = None
+ self.placeholder = placeholder
+ self.validation_types = validation_types or []
+
+ first_entry = self.create_entry_row()
+ self.entries.append(first_entry)
+
+ self.add_button = urwid.Button("+ Add Another", on_press=self.add_entry)
+ add_button_padded = urwid.Padding(self.add_button, left=2, right=2)
+
+ pile_widgets = [first_entry, add_button_padded]
+ urwid.Pile.__init__(self, pile_widgets)
+ FormField.__init__(self, config_key, transform)
+
+ def create_entry_row(self):
+ edit = urwid.Edit("", "")
+ entry_row = urwid.Columns([
+ ('weight', 1, edit),
+ (3, urwid.Button("×", on_press=lambda button: self.remove_entry(button, entry_row))),
+ ])
+ return entry_row
+
+ def remove_entry(self, button, entry_row):
+ if len(self.entries) > 1:
+ self.entries.remove(entry_row)
+ self.contents = [(w, self.options()) for w in self.get_pile_widgets()]
+
+ def add_entry(self, button):
+ new_entry = self.create_entry_row()
+ self.entries.append(new_entry)
+
+ self.contents = [(w, self.options()) for w in self.get_pile_widgets()]
+
+ def get_pile_widgets(self):
+ return self.entries + [urwid.Padding(self.add_button, left=2, right=2)]
+
+ def get_value(self):
+ values = []
+ for entry in self.entries:
+ edit_widget = entry.contents[0][0]
+ value = edit_widget.edit_text.strip()
+ if value:
+ values.append(value)
+ return self.transform(values)
+
+ def validate(self):
+ values = self.get_value()
+ self.error = None
+
+ for validation in self.validation_types:
+ if validation == "required" and not values:
+ self.error = "At least one entry is required"
+ break
+
+ self.error_widget.set_text(("error", self.error or ""))
+ return self.error is None
\ No newline at end of file
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────